Benzene

Getting Started: Benzene over gRPC

Benzene can expose your message handlers as the implementation of a gRPC service, and call other gRPC services back through the same transport-agnostic client surface you use everywhere else. On the server side, @benzene/grpc bridges a @grpc/grpc-js Server into a Benzene middleware pipeline, routing calls of all four RPC shapes — unary, server-streaming, client-streaming, and bidirectional — to the handler whose topic matches. On the client side, @benzene/grpc-client sends unary calls out through that same pipeline model. Both sides share a Benzene-result ↔ gRPC-status mapping, so a handler's BenzeneResult status becomes a gRPC StatusCode (and a benzene-status trailer) on the way out, and is recovered on the way back in.

If you're brand new to Benzene, read Getting Started first — it builds the same kind of service locally on Express in about five minutes. The message handler you write there runs unchanged over gRPC; only the entry point differs, and that's what this guide covers.

TypeScript port. This is the TypeScript port of Benzene. Because Node has no ASP.NET Core, the .NET package's BenzeneInterceptor and Benzene.Grpc.AspNet hosting glue have no analog here: the @grpc/grpc-js Server is the host, and a single useGrpc(...) bridge replaces both — you register the bridge's handlers on the server directly. A few pieces are deliberately not ported: the gRPC health check / reflection services, rich google.rpc.Status error details (the flat benzene-status trailer is ported), and — on the client — non-unary streaming calls. See each package's index.ts "SCOPE" note for the full rationale.

Prerequisites

1. Create the project

mkdir orders-grpc && cd orders-grpc
npm init -y
npm pkg set type=module

Setting type=module makes this an ES-module project, which Benzene's packages require.

2. Install the packages

npm install @benzene/grpc @benzene/core-message-handlers @benzene/results @grpc/grpc-js
# add the client only if this service also calls other gRPC services:
npm install @benzene/grpc-client @benzene/clients

@benzene/grpc is the server bridge; @grpc/grpc-js is the gRPC runtime it wires into (a peer you supply). @benzene/core-message-handlers provides the @message decorator and useMessageHandlers, and @benzene/results provides BenzeneResult. Add @benzene/grpc-client and @benzene/clients only if this service is also a gRPC caller (see step 7).

The core idea in 30 seconds

Benzene separates what your service does from how it's invoked:

@grpcMethod only annotates the class — it does not register it. Registration comes from @message (which self-registers in the handler registry); @grpcMethod's path is then read off each handler to build the method-path → topic route table. So the two decorators always travel together.

3. Define your .proto

Nothing Benzene-specific here — this is an ordinary gRPC service:

syntax = "proto3";
package orders;

service Orders {
  rpc PlaceOrder (PlaceOrderRequest) returns (OrderConfirmation);
}

message PlaceOrderRequest { string customer_id = 1; }
message OrderConfirmation { string order_id = 1; }

Load it into a grpc-js ServiceDefinition the standard way (for example with @grpc/proto-loader); that ServiceDefinition is what you pass to server.addService(...) in step 5. Benzene doesn't replace this step — it plugs handlers into it.

4. Write a message handler

This is where your logic lives — the file you'd carry over verbatim if you later hosted it on Express or Lambda. Two decorators do the wiring:

// src/handlers.ts
import { IBenzeneResultOf } from '@benzene/abstractions';
import { IMessageHandler } from '@benzene/abstractions-message-handlers';
import { message } from '@benzene/core-message-handlers';
import { BenzeneResult } from '@benzene/results';
import { grpcMethod } from '@benzene/grpc';

// Payloads are classes, not interfaces: the runtime recovers the erased request type from its
// constructor (for topic/schema keying), which an interface can't provide.
export class PlaceOrder {
  customerId = '';
}

export class OrderConfirmation {
  orderId = '';
}

@grpcMethod('/orders.Orders/PlaceOrder')
@message('order:place', { requestType: PlaceOrder, responseType: OrderConfirmation })
export class PlaceOrderHandler implements IMessageHandler<PlaceOrder, OrderConfirmation> {
  handleAsync(request: PlaceOrder): Promise<IBenzeneResultOf<OrderConfirmation>> {
    const confirmation = new OrderConfirmation();
    confirmation.orderId = `order-${request.customerId || 'anon'}`;
    return Promise.resolve(BenzeneResult.ok(confirmation));
  }
}

BenzeneResult.ok(...) is the success case; the result carries success/failure status alongside the payload — see Message Result.

5. Wire up the gRPC server

useGrpc(configure) builds the Benzene pipeline (registering the gRPC + baseline services on its own), tags the transport "grpc", and returns a GrpcBenzeneBridge. The bridge's to*Handler(methodPath) methods produce grpc-js handlers you drop straight into server.addService(...) — one per RPC shape:

// src/index.ts
import { Server, ServerCredentials } from '@grpc/grpc-js';
import { useMessageHandlers } from '@benzene/core-message-handlers';
import { useGrpc } from '@benzene/grpc';
import { PlaceOrderHandler } from './handlers.js';
import { OrdersService } from './generated/orders.js'; // your loaded ServiceDefinition

const bridge = useGrpc((pipeline) => useMessageHandlers(pipeline, PlaceOrderHandler));

const server = new Server();
server.addService(OrdersService, {
  placeOrder: bridge.toUnaryHandler('/orders.Orders/PlaceOrder'),
});

server.bindAsync('0.0.0.0:50051', ServerCredentials.createInsecure(), () => {
  console.log('gRPC server listening on :50051');
});

What each step does:

On a match, the bridge runs the pipeline and invokes the grpc-js callback with the response plus a benzene-status trailer. If no Benzene handler owns the called method, the bridge fails the call with gRPC UNIMPLEMENTED — the Node analog of .NET's interceptor falling through to a native service method.

6. Streaming handlers

A streaming handler is an ordinary message handler whose request and/or response type is AsyncIterable<T>. Register it with the matching to*Handler for its shape:

// server-streaming: one request in, a stream of responses out
@grpcMethod('/orders.Orders/Subscribe')
@message('order:subscribe', { requestType: SubscribeRequest })
export class SubscribeHandler
  implements IMessageHandler<SubscribeRequest, AsyncIterable<OrderEvent>>
{
  handleAsync(request: SubscribeRequest): Promise<IBenzeneResultOf<AsyncIterable<OrderEvent>>> {
    return Promise.resolve(BenzeneResult.ok(produce(request)));
  }
}

server.addService(OrdersService, {
  placeOrder: bridge.toUnaryHandler('/orders.Orders/PlaceOrder'),
  subscribe: bridge.toServerStreamingHandler('/orders.Orders/Subscribe'),
  upload:    bridge.toClientStreamingHandler('/orders.Orders/Upload'),   // AsyncIterable<T> in, T out
  chat:      bridge.toBidiStreamingHandler('/orders.Orders/Chat'),       // AsyncIterable<T> both ways
});

One pipeline invocation happens per RPC call, not per stream item. For the response-writing shapes (server-/bidi-streaming) the bridge writes each yielded item to the call, then end()s it with the benzene-status trailer on success; on a non-OK status it emits a gRPC error on the call.

Status mapping

Every handler result's BenzeneResult status is mapped to a gRPC StatusCode and also written verbatim onto a benzene-status response trailer. The DefaultGrpcStatusCodeMapper table:

BenzeneResultStatus gRPC status
ok, ignored, created, accepted, updated, deleted OK
badRequest, validationError INVALID_ARGUMENT
unauthorized UNAUTHENTICATED
forbidden PERMISSION_DENIED
notFound NOT_FOUND
conflict ALREADY_EXISTS
notImplemented UNIMPLEMENTED
serviceUnavailable UNAVAILABLE
tooManyRequests RESOURCE_EXHAUSTED
timeout DEADLINE_EXCEEDED
unexpectedError / anything unrecognized INTERNAL

A non-OK status fails the call with that code; the details carry the joined result errors. Because the benzene-status trailer is always added, a Benzene client can recover the original, more specific status even where several statuses collapse onto the same code (e.g. created/accepted/updated all map to OK) — that's what the client's reverse mapper prefers (see the next step).

7. Calling other gRPC services

@benzene/grpc-client's GrpcBenzeneMessageClient is an IBenzeneMessageClient that sends unary calls out through a Benzene pipeline over a @grpc/grpc-js Client you own. Register a topic → method route for each outbound call, then send by topic:

import { Client, ChannelCredentials } from '@grpc/grpc-js';
import { sendMessageAsync } from '@benzene/clients';
import { GrpcBenzeneMessageClient, GrpcClientRouteRegistry } from '@benzene/grpc-client';

const registry = new GrpcClientRouteRegistry();
registry.add('order:place', '/orders.Orders/PlaceOrder');

const grpcClient = new Client('localhost:50051', ChannelCredentials.createInsecure());
const client = new GrpcBenzeneMessageClient(grpcClient, registry);

const result = await sendMessageAsync(client, 'order:place', { customerId: 'acme' });
if (result.isSuccessful) {
  console.log(result.payload); // { orderId: 'order-acme' }
} else {
  console.error(result.status, result.errors);
}

DI wiring

To resolve the client from a container instead of constructing it by hand, use addGrpcClient:

import { addGrpcClient } from '@benzene/grpc-client';
import { IBenzeneMessageClient } from '@benzene/clients';

addGrpcClient(container, grpcClient, (registry) => {
  registry.add('order:place', '/orders.Orders/PlaceOrder');
});
// later: resolver.getService(IBenzeneMessageClient)

8. Testing

A Benzene message handler is a plain class — the fastest test constructs it and calls handleAsync directly, with no gRPC in the picture:

import { describe, expect, it } from 'vitest';
import { PlaceOrderHandler, PlaceOrder } from '../src/handlers.js';

describe('PlaceOrderHandler', () => {
  it('confirms the order', async () => {
    const request = new PlaceOrder();
    request.customerId = 'acme';

    const result = await new PlaceOrderHandler().handleAsync(request);

    expect(result.isSuccessful).toBe(true);
    expect(result.payload.orderId).toBe('order-acme');
  });
});

To exercise the whole pipeline, useGrpc(...) returns the same bridge you ship, so you can invoke a to*Handler against a stand-in grpc-js call object and assert on the response and the benzene-status trailer. On the client side, new GrpcBenzeneMessageClient(fakeClient, registry) accepts any object with a makeUnaryRequest method, so a fake grpc-js Client lets you drive sendMessageAsync end-to-end without a live connection. See the package tests under test/Benzene.Core.Test/Grpc/ for worked examples of both.

See Also