gRPC & Inter-Service Communication

# ===== SERVER =====
from concurrent import futures
import grpc
import user_pb2 as pb          # generated: messages
import user_pb2_grpc as pb_grpc  # generated: service base classes

class UserService(pb_grpc.UserServiceServicer):
    # Method signature generated FROM the proto: (self, request, context) -> response
    def GetUser(self, request, context):
        if not request.id:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "id is required")  # typed error, sec 14
        # ...real work: query the DB by request.id...
        user = pb.User(id=request.id, email="ada@example.com",
                       full_name="Ada", role=pb.ROLE_ADMIN)
        return pb.GetUserResponse(user=user)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    pb_grpc.add_UserServiceServicer_to_server(UserService(), server)  # wire impl to service
    server.add_insecure_port("[::]:50051")  # insecure = local dev only
    server.start()
    print("gRPC on :50051")
    server.wait_for_termination()

# ===== CLIENT =====
def call_get_user():
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = pb_grpc.UserServiceStub(channel)        # the generated STUB
        try:
            # timeout= is the deadline, sec 13; looks local, runs remote
            resp = stub.GetUser(pb.GetUserRequest(id="u42"), timeout=1.0)
            print("got user:", resp.user.full_name)
        except grpc.RpcError as e:
            print("failed:", e.code(), e.details())     # carries the gRPC status code

if __name__ == "__main__":
    serve()

// ===== SERVER =====
const grpc = require('@grpc/grpc-js')
const protoLoader = require('@grpc/proto-loader')

// Node is the odd one out: it loads the .proto at RUNTIME instead of
// generating source ahead of time. Same contract, later moment, nothing
// about the shape is checked until the process starts.
const def = protoLoader.loadSync('user.proto', { longs: String, enums: String })
const { user } = grpc.loadPackageDefinition(def)

// The method name and its (call, callback) shape come FROM the proto.
function getUser(call, callback) {
  if (!call.request.id) {
    // typed error, sec 14
    return callback({ code: grpc.status.INVALID_ARGUMENT, message: 'id is required' })
  }
  // ...real work: query the DB by call.request.id...
  callback(null, {
    user: {
      id: call.request.id,
      email: 'ada@example.com',
      fullName: 'Ada',      // proto's full_name arrives camelCased
      role: 'ROLE_ADMIN',
    },
  })
}

const server = new grpc.Server()
server.addService(user.v1.UserService.service, { getUser }) // wire impl to service
server.bindAsync('0.0.0.0:50051', grpc.ServerCredentials.createInsecure(), () => {
  console.log('gRPC on :50051') // insecure creds for local dev only
})

// ===== CLIENT =====
function callGetUser() {
  const client = new user.v1.UserService(
    'localhost:50051',
    grpc.credentials.createInsecure(),
  )

  const deadline = new Date(Date.now() + 1000) // always a deadline, sec 13
  client.getUser({ id: 'u42' }, { deadline }, (err, resp) => {
    if (err) {
      // err.code carries the gRPC status code
      return console.error('GetUser failed:', err.code, err.details)
    }
    console.log('got user:', resp.user.fullName)
  })
}

// ===== SERVER =====
import * as grpc from '@grpc/grpc-js'
// ts-proto turns user.proto into real TypeScript at BUILD time, and that
// is the whole reason to prefer it to runtime loading: the compiler now
// knows GetUserRequest has an `id` and nothing else, so a typo in a field
// name fails the build instead of silently sending undefined.
import {
  UserServiceService,
  UserServiceClient,
  Role,
  type UserServiceServer,
  type GetUserResponse,
} from './gen/user'

const userService: UserServiceServer = {
  // The signature is generated FROM the proto: (call, callback)
  getUser(call, callback): void {
    if (!call.request.id) {
      callback({ code: grpc.status.INVALID_ARGUMENT, message: 'id is required' })
      return // typed error, sec 14
    }
    // ...real work: query the DB by call.request.id...
    const response: GetUserResponse = {
      user: {
        id: call.request.id,
        email: 'ada@example.com',
        fullName: 'Ada',
        role: Role.ROLE_ADMIN, // a real enum member, not the string
        tags: [],
        createdAt: 0,
      },
    }
    callback(null, response)
  },
}

const server = new grpc.Server()
server.addService(UserServiceService, userService) // wire impl to service
server.bindAsync('0.0.0.0:50051', grpc.ServerCredentials.createInsecure(), () => {
  console.log('gRPC on :50051')
})

// ===== CLIENT =====
// grpc-js is callback-based. Promisify it once, here, so that every call
// site downstream can just await the RPC like any other async function.
const client = new UserServiceClient('localhost:50051', grpc.credentials.createInsecure())

export function getUser(id: string): Promise<GetUserResponse> {
  return new Promise((resolve, reject) => {
    const deadline = new Date(Date.now() + 1000) // always a deadline, sec 13
    client.getUser({ id }, { deadline }, (err, resp) => {
      if (err) reject(err) // err carries the gRPC status code
      else resolve(resp)
    })
  })
}

// ===== SERVER =====
// protoc-gen-grpc-java emits UserServiceGrpc with a base class to extend.
// Note the shape: the response goes to a StreamObserver instead of being
// returned, which is why unary and streaming look nearly identical here.
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {

    @Override
    public void getUser(GetUserRequest req,
                        StreamObserver<GetUserResponse> responseObserver) {
        if (req.getId().isEmpty()) {
            responseObserver.onError(Status.INVALID_ARGUMENT
                .withDescription("id is required")
                .asRuntimeException()); // typed error, sec 14
            return;
        }

        // ...real work: query the DB by req.getId()...
        User u = User.newBuilder()
            .setId(req.getId())
            .setEmail("ada@example.com")
            .setFullName("Ada")
            .setRole(Role.ROLE_ADMIN)
            .build();

        responseObserver.onNext(GetUserResponse.newBuilder().setUser(u).build());
        // onCompleted() IS the OK trailer. Forget it and the client hangs
        // until its deadline, with no error to explain why.
        responseObserver.onCompleted();
    }
}

Server server = ServerBuilder.forPort(50051)
    .addService(new UserServiceImpl()) // wire the impl to the service
    .build()
    .start();
System.out.println("gRPC on :50051");

// ===== CLIENT =====
// usePlaintext() is the insecure-credentials equivalent: local dev only.
ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
    .usePlaintext()
    .build();

// The BLOCKING stub is the generated one that reads like a local call.
UserServiceGrpc.UserServiceBlockingStub client = UserServiceGrpc.newBlockingStub(channel)
    .withDeadlineAfter(1, TimeUnit.SECONDS); // always a deadline, sec 13

try {
    GetUserResponse resp = client.getUser(
        GetUserRequest.newBuilder().setId("u42").build()); // looks local, runs remote
    System.out.println("got user: " + resp.getUser().getFullName());
} catch (StatusRuntimeException e) {
    System.err.println("GetUser failed: " + e.getStatus().getCode());
}

# proto:  rpc ListUsers(ListUsersRequest) returns (stream User);

# ===== SERVER: a generator, every `yield` sends one message =====
class UserService(pb_grpc.UserServiceServicer):
    def ListUsers(self, request, context):
        for u in query_users(request.filter):  # big result set
            yield u                             # yielding pushes one message down the stream
        # function returning ends the stream cleanly

# ===== CLIENT: the call returns an ITERABLE of responses =====
def list_users(stub):
    responses = stub.ListUsers(pb.ListUsersRequest(filter="active"), timeout=10.0)
    for u in responses:        # iterate until the server closes the stream
        print("user:", u.full_name)

// proto:  rpc ListUsers(ListUsersRequest) returns (stream User);

// ===== SERVER: `call` IS a writable stream; write() per message =====
function listUsers(call) {
  for (const u of queryUsers(call.request.filter)) { // imagine a big result set
    call.write(u) // push one message down the stream
  }
  call.end() // ending the stream closes it cleanly (sends the OK trailer)
}

// ===== CLIENT: the call is a readable stream; listen for 'data' =====
function listUsersClient(client) {
  const deadline = new Date(Date.now() + 10_000)
  const call = client.listUsers({ filter: 'active' }, { deadline })

  call.on('data', (u) => console.log('user:', u.fullName))
  call.on('end', () => console.log('done')) // server closed the stream
  call.on('error', (err) => console.error(err))
}

// proto:  rpc ListUsers(ListUsersRequest) returns (stream User);

// ===== SERVER: the call object is a typed writable stream =====
const userService: UserServiceServer = {
  listUsers(call: ServerWritableStream<ListUsersRequest, User>): void {
    for (const u of queryUsers(call.request.filter)) {
      call.write(u) // push one message down the stream
    }
    call.end() // closes the stream cleanly
  },
}

// ===== CLIENT: `for await` instead of an 'data' handler =====
// A grpc-js call is an async iterable, and that is worth using rather
// than .on('data'): an event handler fires whether or not you are ready
// for the next message, while `for await` only pulls the next one when
// the loop body has finished, which is real back-pressure.
async function listUsers(client: UserServiceClient): Promise<void> {
  const deadline = new Date(Date.now() + 10_000)
  const call = client.listUsers({ filter: 'active' }, { deadline })

  for await (const u of call as AsyncIterable<User>) {
    console.log('user:', u.fullName)
  }
}

// proto:  rpc ListUsers(ListUsersRequest) returns (stream User);

// ===== SERVER: onNext() repeatedly, then onCompleted() once =====
@Override
public void listUsers(ListUsersRequest req, StreamObserver<User> responseObserver) {
    for (User u : queryUsers(req.getFilter())) { // imagine a big result set
        responseObserver.onNext(u); // push one message down the stream
    }
    responseObserver.onCompleted(); // closes the stream cleanly (OK trailer)
}

// ===== CLIENT: the blocking stub hands back an ITERATOR =====
UserServiceBlockingStub client = UserServiceGrpc.newBlockingStub(channel)
    .withDeadlineAfter(10, TimeUnit.SECONDS);

Iterator<User> users = client.listUsers(
    ListUsersRequest.newBuilder().setFilter("active").build());

while (users.hasNext()) { // iterate until the server closes the stream
    System.out.println("user: " + users.next().getFullName());
}

# proto:  rpc UploadEvents(stream Event) returns (UploadSummary);

# ===== SERVER: request_iterator yields incoming messages; return once =====
class UserService(pb_grpc.UserServiceServicer):
    def UploadEvents(self, request_iterator, context):
        count = 0
        for ev in request_iterator:   # iterate the client's stream
            store(ev)
            count += 1
        return pb.UploadSummary(received=count)  # single reply after client closes

# ===== CLIENT: pass an iterable/generator of requests; get one response =====
def upload_events(stub, events):
    def gen():
        for ev in events:
            yield ev
    summary = stub.UploadEvents(gen(), timeout=30.0)  # returns the single summary
    print("server received", summary.received, "events")

// proto:  rpc UploadEvents(stream Event) returns (UploadSummary);

// ===== SERVER: `call` is readable; reply once via the callback =====
function uploadEvents(call, callback) {
  let count = 0
  call.on('data', (ev) => {
    store(ev)
    count++
  })
  call.on('end', () => callback(null, { received: count })) // client done, reply once
  call.on('error', (err) => callback(err))
}

// ===== CLIENT: write() many, then end(); the callback gets the reply =====
function uploadEventsClient(client, events) {
  const deadline = new Date(Date.now() + 30_000)
  const call = client.uploadEvents({ deadline }, (err, summary) => {
    if (err) return console.error(err)
    console.log('server received', summary.received, 'events')
  })

  for (const ev of events) {
    call.write(ev) // push each one up
  }
  call.end() // close our side; the callback above fires with the summary
}

// proto:  rpc UploadEvents(stream Event) returns (UploadSummary);

// ===== SERVER: a typed readable stream + a single callback =====
const userService: UserServiceServer = {
  uploadEvents(
    call: ServerReadableStream<Event, UploadSummary>,
    callback: sendUnaryData<UploadSummary>,
  ): void {
    let count = 0
    call.on('data', (ev: Event) => {
      store(ev)
      count++
    })
    call.on('end', () => callback(null, { received: count }))
  },
}

// ===== CLIENT: promisified, because the reply arrives in a callback =====
// Worth noticing what the Promise buys beyond ergonomics: it forces the
// two failure paths, a stream error and a call error, into one rejection,
// so a caller cannot await the upload and miss a mid-stream failure.
function uploadEvents(client: UserServiceClient, events: Event[]): Promise<UploadSummary> {
  return new Promise((resolve, reject) => {
    const deadline = new Date(Date.now() + 30_000)
    const call = client.uploadEvents({ deadline }, (err, summary) => {
      if (err) reject(err)
      else resolve(summary)
    })
    call.on('error', reject)

    for (const ev of events) call.write(ev)
    call.end() // close our side, await the summary
  })
}

// proto:  rpc UploadEvents(stream Event) returns (UploadSummary);

// ===== SERVER: RETURN an observer; the client's messages arrive on it =====
@Override
public StreamObserver<Event> uploadEvents(StreamObserver<UploadSummary> responseObserver) {
    // The inversion here is the one thing to get used to in Java: the
    // parameter is where you WRITE, and the return value is where the
    // client's messages are READ.
    return new StreamObserver<Event>() {
        private int count = 0;

        @Override public void onNext(Event ev) {
            store(ev);
            count++;
        }

        @Override public void onError(Throwable t) {
            // client vanished mid-upload; nothing to reply to
        }

        @Override public void onCompleted() { // client finished sending
            responseObserver.onNext(
                UploadSummary.newBuilder().setReceived(count).build());
            responseObserver.onCompleted(); // single reply, then close
        }
    };
}

// ===== CLIENT: the ASYNC stub; a blocking stub cannot send a stream =====
UserServiceStub client = UserServiceGrpc.newStub(channel);
CountDownLatch done = new CountDownLatch(1);

StreamObserver<Event> requests = client.uploadEvents(new StreamObserver<UploadSummary>() {
    @Override public void onNext(UploadSummary s) {
        System.out.println("server received " + s.getReceived() + " events");
    }
    @Override public void onError(Throwable t) { done.countDown(); }
    @Override public void onCompleted() { done.countDown(); }
});

for (Event ev : events) {
    requests.onNext(ev); // push each one up
}
requests.onCompleted(); // close our side, await the summary

done.await(30, TimeUnit.SECONDS);

# proto:  rpc Chat(stream ChatMessage) returns (stream ChatMessage);

# ===== SERVER: iterate incoming, yield outgoing, interleaved =====
class UserService(pb_grpc.UserServiceServicer):
    def Chat(self, request_iterator, context):
        for msg in request_iterator:          # read the client's stream
            yield pb.ChatMessage(user="server", text="ack: " + msg.text)  # write back

# ===== CLIENT: pass a request generator; iterate the response stream =====
def chat(stub):
    def outgoing():
        for t in ["hi", "how are you", "bye"]:
            yield pb.ChatMessage(user="ada", text=t)
    responses = stub.Chat(outgoing())   # both directions live at once
    for reply in responses:
        print("<<", reply.text)

# (for true concurrency under load, prefer the async API: grpc.aio)

// proto:  rpc Chat(stream ChatMessage) returns (stream ChatMessage);

// ===== SERVER: `call` is duplex, read and write on the same object =====
function chat(call) {
  call.on('data', (msg) => {
    // echo back (or broadcast to a room, etc.), can write anytime, any number
    call.write({ user: 'server', text: 'ack: ' + msg.text })
  })
  call.on('end', () => call.end()) // client closed its send side
}

// ===== CLIENT: one duplex stream; write and listen independently =====
function chatClient(client) {
  const call = client.chat()

  call.on('data', (msg) => console.log('<<', msg.text)) // concurrent receiver

  for (const text of ['hi', 'how are you', 'bye']) {
    call.write({ user: 'ada', text })
  }
  call.end() // signal we're done sending; 'data' keeps firing for the rest
}

// proto:  rpc Chat(stream ChatMessage) returns (stream ChatMessage);

// ===== SERVER: a typed duplex stream =====
const userService: UserServiceServer = {
  chat(call: ServerDuplexStream<ChatMessage, ChatMessage>): void {
    call.on('data', (msg: ChatMessage) => {
      call.write({ user: 'server', text: 'ack: ' + msg.text })
    })
    call.on('end', () => call.end())
  },
}

// ===== CLIENT: no second thread needed =====
// Compare this with the Go tab, which spawns a goroutine to receive while
// the main function sends. Node needs no such thing: writes go out
// immediately and 'data' fires on the event loop, so one thread is
// already doing both directions. The trade is that nothing ever blocks,
// so there is no natural place where back-pressure makes itself felt,
// you have to watch `call.write()`'s return value for that.
function chat(client: UserServiceClient): void {
  const call = client.chat()

  call.on('data', (msg: ChatMessage) => console.log('<<', msg.text))

  for (const text of ['hi', 'how are you', 'bye']) {
    call.write({ user: 'ada', text })
  }
  call.end() // signal we're done sending; the receiver drains the rest
}

// proto:  rpc Chat(stream ChatMessage) returns (stream ChatMessage);

// ===== SERVER: read from the returned observer, write to the parameter =====
@Override
public StreamObserver<ChatMessage> chat(StreamObserver<ChatMessage> responseObserver) {
    return new StreamObserver<ChatMessage>() {
        @Override public void onNext(ChatMessage msg) {
            // interleaved: reply as each message arrives, any number of times
            responseObserver.onNext(ChatMessage.newBuilder()
                .setUser("server")
                .setText("ack: " + msg.getText())
                .build());
        }

        @Override public void onError(Throwable t) { }

        @Override public void onCompleted() {
            responseObserver.onCompleted(); // client closed its send side
        }
    };
}

// ===== CLIENT: the async stub; both directions live at once =====
UserServiceStub client = UserServiceGrpc.newStub(channel);

StreamObserver<ChatMessage> outgoing = client.chat(new StreamObserver<ChatMessage>() {
    @Override public void onNext(ChatMessage in) {
        System.out.println("<< " + in.getText()); // concurrent receiver
    }
    @Override public void onError(Throwable t) { }
    @Override public void onCompleted() { }
});

for (String t : List.of("hi", "how are you", "bye")) {
    outgoing.onNext(ChatMessage.newBuilder().setUser("ada").setText(t).build());
}
outgoing.onCompleted(); // signal we're done sending; the receiver drains the rest

# Install the tooling once:
#   pip install grpcio grpcio-tools

# Generate messages AND service stubs in one command:
#   python -m grpc_tools.protoc \
#       -I. \
#       --python_out=. \
#       --grpc_python_out=. \
#       user.proto

# Produces:
#   user_pb2.py, message classes
#   user_pb2_grpc.py, UserServiceStub (client) + UserServiceServicer (to implement)

# Then import both in your code:
#   import user_pb2 as pb
#   import user_pb2_grpc as pb_grpc

// Node does NOT generate source by default. Install the runtime pair:
//   npm install @grpc/grpc-js @grpc/proto-loader

// There is no build step: proto-loader reads user.proto at STARTUP and
// builds the client and server constructors in memory.
//   const def = protoLoader.loadSync('user.proto')
//   const { user } = grpc.loadPackageDefinition(def)

// Produces: nothing on disk. The service is discovered at runtime as
//   user.v1.UserService  (client constructor)
//   user.v1.UserService.service  (the descriptor you addService() with)

// The trade is real and worth stating: no generated files to review, and
// no check at all that your handler object matches the contract. A typo
// in a method name is simply an unimplemented RPC at runtime.

// For TypeScript you DO want generated source, so that the contract is a
// compile-time thing again. Install protoc plus a TS plugin:
//   npm install -D ts-proto
//   npm install @grpc/grpc-js

// Generate typed messages AND grpc-js service definitions:
//   protoc \
//     --plugin=./node_modules/.bin/protoc-gen-ts_proto \
//     --ts_proto_out=./src/gen \
//     --ts_proto_opt=outputServices=grpc-js,esModuleInterop=true \
//     user.proto

// Produces:
//   src/gen/user.ts, interfaces for each message, a real TS enum for Role,
//                    UserServiceClient (stub) + UserServiceServer (to implement)

// Then import it like any other module:
//   import { UserServiceClient, type UserServiceServer } from './gen/user'

// Java generates at BUILD time, wired into the build tool rather than run
// by hand. With Gradle and the protobuf plugin:
//
//   plugins { id 'com.google.protobuf' version '0.9.4' }
//
//   protobuf {
//     protoc { artifact = 'com.google.protobuf:protoc:3.25.3' }
//     plugins {
//       grpc { artifact = 'io.grpc:protoc-gen-grpc-java:1.62.2' }
//     }
//     generateProtoTasks { all()*.plugins { grpc {} } }
//   }
//
// Drop user.proto in src/main/proto/ and `./gradlew build` regenerates it.

// Produces (in build/generated/source/proto/main):
//   User.java, GetUserRequest.java, ... immutable classes + Builders
//   UserServiceGrpc.java, UserServiceImplBase (to extend)
//                       + newBlockingStub / newStub / newFutureStub

// Then import the generated package:
//   import com.example.gen.userv1.UserServiceGrpc;

# ===== CLIENT: attach a deadline (timeout=) + metadata =====
metadata = (
    ("authorization", f"Bearer {token}"),  # auth as metadata, sec 16
    ("x-request-id", req_id),               # trace id for correlation
)
try:
    resp = stub.GetUser(pb.GetUserRequest(id="u42"),
                        timeout=1.0,         # the deadline, in seconds
                        metadata=metadata)
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print("call timed out")             # a network-shaped failure

# ===== SERVER: read metadata, check for cancellation =====
class UserService(pb_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        md = dict(context.invocation_metadata())
        auth = md.get("authorization")       # verify token (or in an interceptor, sec 15)

        if not context.is_active():          # client gone / deadline passed?
            return pb.GetUserResponse()      # abandon the work
        return self.lookup(request.id)

// ===== CLIENT: attach a deadline + metadata =====
const metadata = new grpc.Metadata()
metadata.set('authorization', `Bearer ${token}`) // auth as metadata, sec 16
metadata.set('x-request-id', reqId)              // trace id for correlation

// A deadline in grpc-js is an ABSOLUTE Date, not a duration. That is the
// right shape: as the call is retried or forwarded, everyone downstream
// inherits the same wall-clock instant rather than a fresh N seconds.
const deadline = new Date(Date.now() + 1000)

client.getUser({ id: 'u42' }, metadata, { deadline }, (err, resp) => {
  if (err?.code === grpc.status.DEADLINE_EXCEEDED) {
    console.log('call timed out') // a network-shaped failure a local call never had
  }
})

// ===== SERVER: read metadata, notice cancellation =====
function getUser(call, callback) {
  const auth = call.metadata.get('authorization')[0] // verify here or in an interceptor, sec 15

  if (call.cancelled) {
    return // client gone / deadline passed; abandon the work
  }
  call.on('cancelled', () => abortWork()) // fires if they hang up mid-call
  lookup(call.request.id).then((u) => callback(null, u))
}

// ===== CLIENT: a wrapper, so no call site can forget the deadline =====
// This is the point worth making in TypeScript: a deadline you must
// remember to pass is a deadline somebody will eventually omit. Make the
// parameter required and the omission stops compiling.
function callWithDeadline<Req, Res>(
  method: (req: Req, md: grpc.Metadata, opts: grpc.CallOptions,
           cb: (e: grpc.ServiceError | null, r: Res) => void) => void,
  req: Req,
  timeoutMs: number, // required, not optional
): Promise<Res> {
  const md = new grpc.Metadata()
  md.set('authorization', `Bearer ${token}`) // auth as metadata, sec 16
  md.set('x-request-id', reqId)              // trace id for correlation

  return new Promise((resolve, reject) => {
    const deadline = new Date(Date.now() + timeoutMs) // absolute deadline
    method(req, md, { deadline }, (err, res) => {
      if (err?.code === grpc.status.DEADLINE_EXCEEDED) {
        reject(new Error('call timed out'))
      } else if (err) {
        reject(err)
      } else {
        resolve(res)
      }
    })
  })
}

// ===== SERVER: read metadata, respect cancellation =====
const userService: UserServiceServer = {
  getUser(call, callback): void {
    const auth = call.metadata.get('authorization')[0]

    // An AbortSignal is how the cancellation is threaded onward, the
    // Node equivalent of passing Go's ctx down to the DB driver.
    const ac = new AbortController()
    call.on('cancelled', () => ac.abort())

    lookup(call.request.id, ac.signal).then(
      (user) => callback(null, { user }),
      (err) => callback(err),
    )
  },
}

// ===== CLIENT: attach a deadline + metadata =====
Metadata headers = new Metadata();
Metadata.Key<String> AUTH =
    Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
Metadata.Key<String> REQ_ID =
    Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER);
headers.put(AUTH, "Bearer " + token);  // auth travels as metadata, sec 16
headers.put(REQ_ID, reqId);            // trace id for correlation

UserServiceBlockingStub stub = UserServiceGrpc.newBlockingStub(channel)
    .withDeadlineAfter(1, TimeUnit.SECONDS) // absolute deadline for the call
    .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers));

try {
    GetUserResponse resp = stub.getUser(GetUserRequest.newBuilder().setId("u42").build());
} catch (StatusRuntimeException e) {
    if (e.getStatus().getCode() == Status.Code.DEADLINE_EXCEEDED) {
        System.out.println("call timed out"); // a network-shaped failure
    }
}

// ===== SERVER: read metadata, respect the inherited deadline =====
// Headers are read in an interceptor (sec 15) and handed to the method
// through a Context key, which is Java's context: a per-call, immutable
// key-value scope that also carries the deadline and cancellation.
@Override
public void getUser(GetUserRequest req, StreamObserver<GetUserResponse> responseObserver) {
    String auth = AUTH_CONTEXT_KEY.get(); // put there by the auth interceptor

    Context ctx = Context.current();
    if (ctx.isCancelled()) { // client gone / deadline passed?
        responseObserver.onError(Status.CANCELLED.asRuntimeException());
        return; // abandon the work
    }
    // Pass ctx down so slow work is abandoned automatically:
    //   ctx.call(() -> db.lookup(req.getId()));
    responseObserver.onNext(lookup(req.getId()));
    responseObserver.onCompleted();
}

import grpc

# SERVER: abort with a code + message (or set_code/set_details then return).
class UserService(pb_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        if not request.id:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "id is required")
        user = self.db.find(request.id)
        if user is None:
            context.abort(grpc.StatusCode.NOT_FOUND, f"no user with id {request.id!r}")
        return pb.GetUserResponse(user=user)

# CLIENT: catch RpcError and switch on .code()
try:
    resp = stub.GetUser(req, timeout=1.0)
except grpc.RpcError as e:
    if e.code() == grpc.StatusCode.NOT_FOUND:
        ...   # expected, show "not found"
    elif e.code() == grpc.StatusCode.UNAVAILABLE:
        ...   # transient, retry with backoff, sec 17
    else:
        print("unexpected:", e.code(), e.details())

const grpc = require('@grpc/grpc-js')

// SERVER: pass a status object to the callback, not a bare Error.
function getUser(call, callback) {
  if (!call.request.id) {
    return callback({ code: grpc.status.INVALID_ARGUMENT, message: 'id is required' })
  }
  const user = db.find(call.request.id)
  if (!user) {
    return callback({
      code: grpc.status.NOT_FOUND,
      message: `no user with id "${call.request.id}"`,
    })
  }
  callback(null, { user })
}

// Throwing a plain Error instead reaches the client as UNKNOWN with the
// message swallowed, which is the single most common grpc-js mistake.

// CLIENT: inspect err.code to decide what to do.
client.getUser(req, (err, resp) => {
  if (err) {
    switch (err.code) {
      case grpc.status.NOT_FOUND:    break // expected, show "not found" in UI
      case grpc.status.UNAVAILABLE:  break // transient, retry with backoff, sec 17
      default: console.error('unexpected:', err.code, err.details)
    }
  }
})

import * as grpc from '@grpc/grpc-js'

// A narrow, named union of the codes this service actually returns is
// worth the four lines: `switch` then gets exhaustiveness checking, so
// adding a code to the server without handling it on the client fails
// the build instead of falling through to the default branch in prod.
type UserServiceCode =
  | grpc.status.NOT_FOUND
  | grpc.status.INVALID_ARGUMENT
  | grpc.status.UNAVAILABLE

// SERVER: a helper, so every error in the service has the same shape.
function fail(code: grpc.status, message: string): grpc.ServiceError {
  return Object.assign(new Error(message), { code, details: message }) as grpc.ServiceError
}

const userService: UserServiceServer = {
  getUser(call, callback): void {
    if (!call.request.id) {
      callback(fail(grpc.status.INVALID_ARGUMENT, 'id is required'))
      return
    }
    const user = db.find(call.request.id)
    if (!user) {
      callback(fail(grpc.status.NOT_FOUND, `no user with id "${call.request.id}"`))
      return
    }
    callback(null, { user })
  },
}

// CLIENT: switch on the typed code.
function handle(err: grpc.ServiceError): void {
  switch (err.code as UserServiceCode) {
    case grpc.status.NOT_FOUND:       break // expected, show "not found"
    case grpc.status.UNAVAILABLE:     break // transient, retry with backoff, sec 17
    case grpc.status.INVALID_ARGUMENT: break // our bug, fix the call site
  }
}

// SERVER: fail the observer with a typed Status, never a bare exception.
@Override
public void getUser(GetUserRequest req, StreamObserver<GetUserResponse> responseObserver) {
    if (req.getId().isEmpty()) {
        responseObserver.onError(Status.INVALID_ARGUMENT
            .withDescription("id is required")
            .asRuntimeException());
        return;
    }

    Optional<User> user = db.find(req.getId());
    if (user.isEmpty()) {
        responseObserver.onError(Status.NOT_FOUND
            .withDescription("no user with id \"" + req.getId() + "\"")
            .asRuntimeException());
        return;
    }

    responseObserver.onNext(GetUserResponse.newBuilder().setUser(user.get()).build());
    responseObserver.onCompleted();
}

// Any OTHER exception escaping the method becomes UNKNOWN with no
// description at all, which tells the client nothing. Catch and map.

// CLIENT: catch StatusRuntimeException and switch on the code.
try {
    GetUserResponse resp = client.getUser(req);
} catch (StatusRuntimeException e) {
    switch (e.getStatus().getCode()) {
        case NOT_FOUND   -> { }  // expected, show "not found" in UI
        case UNAVAILABLE -> { }  // transient, retry with backoff, sec 17
        default -> System.err.printf("unexpected: %s: %s%n",
            e.getStatus().getCode(), e.getStatus().getDescription());
    }
}

import grpc

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        md = dict(handler_call_details.invocation_metadata)
        token = md.get("authorization", "")
        if not valid(token):
            # short-circuit: abort before the handler ever runs
            def deny(request, context):
                context.abort(grpc.StatusCode.UNAUTHENTICATED, "missing or invalid token")
            return grpc.unary_unary_rpc_method_handler(deny)
        return continuation(handler_call_details)  # proceed to next / handler

# Register interceptors when building the server (applied in order):
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[AuthInterceptor()],
)

# Client-side interceptors also exist (e.g. to inject auth on every call):
#   channel = grpc.intercept_channel(base_channel, MyClientInterceptor())

// grpc-js has no first-class server interceptor API, so the usual shape
// is a wrapper applied to each handler when the service is registered.
// Same idea as chapter 06's middleware: wrap, don't call from inside.
function withAuth(handler) {
  return (call, callback) => {
    const token = call.metadata.get('authorization')[0]
    if (!token || !valid(token)) {
      // short-circuit: the real handler never runs
      return callback({
        code: grpc.status.UNAUTHENTICATED,
        message: 'missing or invalid token',
      })
    }
    call.user = parse(token) // attach the verified identity for the handler
    return handler(call, callback)
  }
}

function withLogging(handler, name) {
  return (call, callback) => {
    const start = Date.now()
    handler(call, (err, resp) => {
      console.log(`${name} took ${Date.now() - start}ms -> ${err ? err.code : 'OK'}`)
      callback(err, resp)
    })
  }
}

// Register the chain when building the service (outermost listed first):
server.addService(user.v1.UserService.service, {
  getUser: withLogging(withAuth(getUser), 'GetUser'),
})

// Typing the wrapper is what keeps a chain honest: `withAuth` returns the
// same handler type it took, so wrappers compose in any order and a
// wrapper that forgets to call the handler fails to type-check.
type Handler<Req, Res> = (
  call: grpc.ServerUnaryCall<Req, Res>,
  callback: grpc.sendUnaryData<Res>,
) => void

function withAuth<Req, Res>(handler: Handler<Req, Res>): Handler<Req, Res> {
  return (call, callback) => {
    const token = call.metadata.get('authorization')[0]
    if (typeof token !== 'string' || !valid(token)) {
      callback({
        code: grpc.status.UNAUTHENTICATED,
        message: 'missing or invalid token',
      } as grpc.ServiceError)
      return // short-circuit before the handler ever runs
    }
    return handler(call, callback) // call the next link / the real handler
  }
}

function withLogging<Req, Res>(name: string, handler: Handler<Req, Res>): Handler<Req, Res> {
  return (call, callback) => {
    const start = Date.now()
    handler(call, (err, resp) => {
      console.log(`${name} took ${Date.now() - start}ms -> ${err?.code ?? 'OK'}`)
      callback(err, resp)
    })
  }
}

// Outermost listed first, exactly as in the Go tab:
server.addService(UserServiceService, {
  getUser: withLogging('GetUser', withAuth(getUser)),
})

// A ServerInterceptor is the real thing in Java, and its shape is fixed
// by the framework, like Go's.
public class AuthInterceptor implements ServerInterceptor {
    static final Context.Key<User> USER = Context.key("user");
    static final Metadata.Key<String> AUTH =
        Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call, Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {

        String token = headers.get(AUTH);
        if (token == null || !valid(token)) {
            call.close(Status.UNAUTHENTICATED
                .withDescription("missing or invalid token"), new Metadata());
            // short-circuit: a no-op listener, the handler never runs
            return new ServerCall.Listener<>() { };
        }

        // Attach the verified identity for the handler to read from Context
        Context ctx = Context.current().withValue(USER, parse(token));
        return Contexts.interceptCall(ctx, call, headers, next);
    }
}

public class LoggingInterceptor implements ServerInterceptor {
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call, Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {
        long start = System.nanoTime();
        return next.startCall(new SimpleForwardingServerCall<>(call) {
            @Override public void close(Status status, Metadata trailers) {
                System.out.printf("%s took %dms -> %s%n",
                    call.getMethodDescriptor().getFullMethodName(),
                    (System.nanoTime() - start) / 1_000_000, status.getCode());
                super.close(status, trailers);
            }
        }, headers);
    }
}

// Register the chain when building the server. Note the ordering trap:
// interceptors run in REVERSE registration order, so the one listed last
// is the outermost. Spelling the chain out in one call keeps it visible.
Server server = ServerBuilder.forPort(50051)
    .addService(ServerInterceptors.intercept(new UserServiceImpl(),
        new AuthInterceptor(), new LoggingInterceptor())) // logging outermost
    .build();

# ===== SERVER over TLS =====
with open("server.key", "rb") as k, open("server.crt", "rb") as c:
    creds = grpc.ssl_server_credentials([(k.read(), c.read())])
server.add_secure_port("[::]:443", creds)   # encrypted port

# ===== CLIENT over TLS =====
with open("ca.crt", "rb") as f:
    channel_creds = grpc.ssl_channel_credentials(root_certificates=f.read())

# ===== Per-call token, composed with the channel credentials =====
class TokenAuth(grpc.AuthMetadataPlugin):
    def __init__(self, token): self.token = token
    def __call__(self, context, callback):
        callback((("authorization", f"Bearer {self.token}"),), None)  # adds metadata per call

call_creds = grpc.metadata_call_credentials(TokenAuth(token))
composite  = grpc.composite_channel_credentials(channel_creds, call_creds)
channel    = grpc.secure_channel("api.example.com:443", composite)

const fs = require('node:fs')
const grpc = require('@grpc/grpc-js')

// ===== SERVER over TLS =====
const serverCreds = grpc.ServerCredentials.createSsl(
  null, // no client CA -> TLS, not mTLS
  [{
    private_key: fs.readFileSync('server.key'),
    cert_chain: fs.readFileSync('server.crt'),
  }],
)
server.bindAsync('0.0.0.0:443', serverCreds, () => {}) // every connection encrypted

// ===== CLIENT over TLS =====
const channelCreds = grpc.credentials.createSsl(fs.readFileSync('ca.crt')) // trust this CA

// ===== Per-call token, composed with the channel credentials =====
const callCreds = grpc.credentials.createFromMetadataGenerator((_params, cb) => {
  const md = new grpc.Metadata()
  md.set('authorization', `Bearer ${token}`) // a fresh token on every call
  cb(null, md)
})

const combined = grpc.credentials.combineChannelCredentials(channelCreds, callCreds)
const client = new user.v1.UserService('api.example.com:443', combined)

// grpc-js refuses to attach call credentials to an insecure channel, the
// same guard as Go's RequireTransportSecurity() returning true.

import * as fs from 'node:fs'
import * as grpc from '@grpc/grpc-js'

// ===== SERVER over TLS =====
const serverCreds = grpc.ServerCredentials.createSsl(
  null,
  [{
    private_key: fs.readFileSync('server.key'),
    cert_chain: fs.readFileSync('server.crt'),
  }],
  false, // checkClientCertificate: flip to true (with a CA above) for mTLS
)

// ===== CLIENT over TLS + a per-call token =====
// The generator runs per call rather than once, and that is the point:
// a token read once at startup is a token that expires in production at
// 3am. Fetch it here, from a cache that refreshes it.
function tokenCredentials(getToken: () => Promise<string>): grpc.CallCredentials {
  return grpc.credentials.createFromMetadataGenerator((_params, cb) => {
    getToken().then(
      (token) => {
        const md = new grpc.Metadata()
        md.set('authorization', `Bearer ${token}`)
        cb(null, md)
      },
      (err) => cb(err),
    )
  })
}

const channelCreds = grpc.credentials.createSsl(fs.readFileSync('ca.crt'))
const client = new UserServiceClient(
  'api.example.com:443',
  grpc.credentials.combineChannelCredentials(channelCreds, tokenCredentials(getToken)),
)

// ===== SERVER over TLS =====
Server server = NettyServerBuilder.forPort(443)
    .useTransportSecurity(new File("server.crt"), new File("server.key"))
    .addService(new UserServiceImpl())
    .build();  // every connection is now encrypted

// For mTLS, demand a client cert signed by a CA you trust:
//   .sslContext(GrpcSslContexts.forServer(crt, key)
//       .trustManager(new File("ca.crt"))
//       .clientAuth(ClientAuth.REQUIRE)
//       .build())

// ===== CLIENT over TLS =====
ManagedChannel channel = NettyChannelBuilder.forAddress("api.example.com", 443)
    .sslContext(GrpcSslContexts.forClient()
        .trustManager(new File("ca.crt")) // trust this CA
        .build())
    .build();

// ===== Per-call token (combine with TLS) =====
// CallCredentials is applied per call, so a refreshed token is picked up
// without rebuilding the channel.
CallCredentials bearer = new CallCredentials() {
    @Override
    public void applyRequestMetadata(RequestInfo info, Executor exec, MetadataApplier applier) {
        exec.execute(() -> {
            Metadata md = new Metadata();
            md.put(Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER),
                   "Bearer " + tokenSupplier.get());
            applier.apply(md);
        });
    }
    @Override public void thisUsesUnstableApi() { }
};

UserServiceBlockingStub stub = UserServiceGrpc.newBlockingStub(channel)
    .withCallCredentials(bearer); // refuses to travel over a plaintext channel