nice-grpc-server-middleware-terminator

import {createServer, ServiceImplementation, ServerError, Status, CallContext} from 'nice-grpc'; import { createTerminatorMiddleware, TerminatorContext } from 'nice-grpc-server-middleware-terminator'; import { AbortController } from 'abort-controller-x'; import * as path from 'path'; import { loadSync } from '@grpc/proto-loader'; import { GrpcObject, PackageDefinition, UntypedServiceImplementation } from '@grpc/grpc-js'; interface ExampleRequest { name: string; } interface ExampleResponse { message: string; } // 1. Define your Protobuf service (example.proto) // syntax = "proto3"; // package example; // service ExampleService { // rpc SayHello (ExampleRequest) returns (ExampleResponse) {} // rpc StreamHello (ExampleRequest) returns (stream ExampleResponse) {} // } const PROTO_PATH = path.resolve(__dirname, 'example.proto'); const packageDefinition: PackageDefinition = loadSync(PROTO_PATH); const grpcObject: GrpcObject = {} as GrpcObject; // Simplified for example, normally loads via @grpc/proto-loader // Simulate the generated service definition const ExampleServiceDefinition = { name: 'example.ExampleService', fullName: 'example.ExampleService', methods: { SayHello: { path: '/example.ExampleService/SayHello', requestStream: false, responseStream: false, requestType: {}, // Simplified responseType: {}, // Simplified responseDeserialize: (b: Buffer) => ({ message: b.toString() }), requestSerialize: (m: ExampleRequest) => Buffer.from(m.name), }, StreamHello: { path: '/example.ExampleService/StreamHello', requestStream: false, responseStream: true, requestType: {}, // Simplified responseType: {}, // Simplified responseDeserialize: (b: Buffer) => ({ message: b.toString() }), requestSerialize: (m: ExampleRequest) => Buffer.from(m.name), } } }; type MyCallContext = CallContext & TerminatorContext; const exampleServiceImpl: ServiceImplementation<typeof ExampleServiceDefinition, MyCallContext> = { async SayHello(request: ExampleRequest, context: MyCallContext): Promise<ExampleResponse> { console.log(`[Server] Received SayHello from ${request.name}`); return { message: `Hello, ${request.name}!` }; }, async *StreamHello(request: ExampleRequest, context: MyCallContext): AsyncIterable<ExampleResponse> { console.log(`[Server] Received StreamHello from ${request.name}. Starting stream...`); let counter = 0; while (true) { try { // IMPORTANT: Check context.signal.aborted regularly in long-running operations // to respond to termination requests. if (context.signal.aborted) { console.log(`[Server] StreamHello for ${request.name} aborted due to server shutdown.`); // This is generally handled by the middleware, but explicit checks are good. throw new ServerError(Status.UNAVAILABLE, 'Server shutting down'); } yield { message: `Stream Hello ${counter++} to ${request.name}` }; await new Promise(resolve => setTimeout(resolve, 500)); // Simulate work } catch (error) { if (error instanceof ServerError && error.code === Status.UNAVAILABLE) { console.log(`[Server] Explicitly caught ServerError for ${request.name}.`); break; // Exit the stream gracefully } console.error(`[Server] Stream error for ${request.name}:`, error); throw error; } } }, }; async function runServer() { const server = createServer() .use(createTerminatorMiddleware()) .add(ExampleServiceDefinition, exampleServiceImpl as UntypedServiceImplementation); const port = 50051; await server.listen(`0.0.0.0:${port}`); console.log(`gRPC server listening on port ${port}`); // Simulate a client making a long-running streaming call // (In a real scenario, this would be a separate client process) async function simulateClient() { const client = createServer().createClient(ExampleServiceDefinition, `0.0.0.0:${port}`); console.log('[Client] Initiating SayHello...'); const unaryResponse = await client.SayHello({ name: 'Alice' }); console.log('[Client] SayHello response:', unaryResponse.message); console.log('[Client] Initiating StreamHello...'); try { for await (const response of client.StreamHello({ name: 'Bob' })) { console.log('[Client] StreamHello response:', response.message); // Simulate client processing for a bit before server shutdown await new Promise(resolve => setTimeout(resolve, 300)); } } catch (error) { if (error instanceof ServerError && error.code === Status.UNAVAILABLE) { console.warn('[Client] Stream terminated by server during shutdown as expected.'); } else { console.error('[Client] StreamHello error:', error); } } } // Start the simulated client call after a short delay setTimeout(simulateClient, 1000); // Simulate graceful shutdown after some time setTimeout(async () => { console.log('\n[Server] Initiating graceful shutdown...'); await server.shutdown(); console.log('[Server] Server shut down gracefully.'); }, 5000); } runServer().catch(console.error);