ProtoBus

RabbitMQ-native microservices for TypeScript, with Protocol Buffers on the wire.

Define a service in a .proto file, implement it as a class, and call it from anywhere on the bus as if it were local. ProtoBus turns each service into one durable RabbitMQ queue with N processes competing for it — so load balancing, failover, backpressure, retries and dead-lettering are the broker's, not JavaScript's.

It is deliberately not transport-agnostic. There is no pluggable-transport abstraction to keep RabbitMQ's features out of reach.

Ports: protobus-py (Python, stable) · protobus-go (Go, experimental). The .proto files are the contract, so they interoperate.

Install

npm install protobus

You also need a RabbitMQ 3.8+ broker:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management-alpine

Quick start

Four steps to a working RPC. Every snippet here is compiled and executed against a real broker by CI on each commit.

1. Describe the service

// proto/Calculator.proto
syntax = "proto3";
package Calculator;

message AddRequest {
    int32 a = 1;
    int32 b = 2;
}

message AddResponse {
    int32 result = 1;
}

service Math {
    rpc add(Calculator.AddRequest) returns(Calculator.AddResponse);
}

Package plus service name is the service's name on the bus: Calculator.Math.

2. Implement it

// src/calculator-service.ts
import { RunnableService } from 'protobus';

export class CalculatorService extends RunnableService {
    public get ServiceName(): string { return 'Calculator.Math'; }

    async add(request: { a: number; b: number }): Promise<{ result: number }> {
        return { result: request.a + request.b };
    }
}

3. Run it

// src/server.ts
import { Context, RunnableService } from 'protobus';
import { CalculatorService } from './calculator-service';

async function main() {
    const context = new Context();
    await context.init(process.env.AMQP_URL || 'amqp://localhost', ['./proto']);

    await RunnableService.start(context, CalculatorService);
    console.log('Calculator.Math is up');
}

main().catch((error) => { console.error(error); process.exit(1); });

RunnableService.start handles SIGINT/SIGTERM, drains in-flight messages on shutdown, and exits non-zero if startup fails.

4. Call it

// src/client.ts
import { Context, ServiceProxy } from 'protobus';

interface CalculatorMath {
    add(request: { a: number; b: number }): Promise<{ result: number }>;
}

async function main() {
    const context = new Context();
    await context.init(process.env.AMQP_URL || 'amqp://localhost', ['./proto']);

    const calculator = new ServiceProxy(context, 'Calculator.Math') as ServiceProxy & CalculatorMath;
    await calculator.init();

    const response = await calculator.add({ a: 5, b: 3 });
    console.log(`5 + 3 = ${response.result}`);

    // A client must close, or the open AMQP socket keeps the process alive.
    await context.connection.disconnect();
}

main().catch((error) => { console.error(error); process.exit(1); });

$ npx tsx src/client.ts
5 + 3 = 8

Full walkthrough, including events and the project layout: Getting Started.

Why ProtoBus

RabbitMQ only, on purpose

ProtoBus is built for one broker, so the things a broker is good at stay in the broker instead of being reimplemented above it:

| Concern | Where it lives | |---|---| | Load balancing | competing consumers on one queue | | Routing | topic exchange bindings (REQUEST.<Service>.*) | | Redelivery on consumer loss | late ack — an unacked delivery returns to the queue | | Retry delay | the retry queue's x-message-ttl, drained by DLX | | Persistence | durable queues, persistent messages | | Dead letters | a real <Service>.DLQ | | Priority | native queue priorities |

A request goes publisher → exchange → queue → consumer. Nothing tracks live instances, so nothing holds a stale one, and a consumer that dies mid-request leaves its delivery unacked for the next consumer to take.

The cost of this is written down rather than glossed over — read Delivery Guarantees before you rely on any of it.

If you may need to swap RabbitMQ for another broker, use a transport-agnostic framework instead. That is a real feature and protobus does not have it.

Protocol Buffers, not JSON

• Smaller on the wire — binary rather than text.
• Contract-first — a .proto file is the interface between teams, and generated types fail the build when the two drift apart.
• Versioning by field number — adding a field does not break an old peer.

The cost: a code-generation step, and no ad-hoc objects.

Cheap to port to another language

Three runtime dependencies, and the messaging behaviour that would be hardest to reimplement — queueing, consumer distribution, retry delays, dead-lettering — is RabbitMQ's, not protobus's. A port swaps the AMQP client and implements a small application protocol on top: three envelope messages, a routing-key scheme, an error encoding, and the streaming headers, all documented in Message Flow. That is real work, but it is bounded, and it is not service discovery or failure recovery.

protobus-py and protobus-go are the existing ports. Note that the cross-language test is excluded from CI — interoperability is checked by hand.

The longer version: Why ProtoBus.

Custom types

Protobuf's scalars do not cover everything. Register a custom type and it becomes usable as a field type in your schemas, encoded and decoded transparently:

import { Context, ICustomType } from 'protobus';

const UuidType: ICustomType<string> = {
    name: 'uuid',                 // how it is written in the .proto
    wireType: 'string',           // how it travels
    tsType: 'string',             // what generated types call it
    encode: (value: string) => value,
    decode: (data: string) => data,
};

async function main() {
    const context = new Context();

    // Register before init(): init() parses your .proto files, and a schema
    // using `uuid` cannot be parsed until the type exists.
    context.factory.registerType(UuidType);

    await context.init('amqp://localhost', ['./proto']);
}

// The schema MUST declare syntax = "proto3" or protobufjs rejects the
// custom type with: illegal token 'uuid'
syntax = "proto3";
package Accounts;

message Account {
    uuid id = 1;
}

BigIntType and TimestampType ship with the library and are already registered; registering either again is a no-op. Details, and the rules that make this work: Custom Types.

CLI

npx protobus generate               # .proto -> TypeScript types
npx protobus generate:service Name  # a runnable service stub
npx protobus init                   # print project setup instructions

Configured from package.json:

{
  "protobus": {
    "protoDir": "./proto",
    "typesOutput": "./common/types/proto.ts",
    "servicesDir": "./services"
  }
}

All three keys are optional; the defaults above are what the CLI uses. CLI reference.

Documentation

Full index: docs/

| Start | | |---|---| | Getting Started | zero to a working RPC, plus events | | Schema Design | writing the .proto that is your contract | | Events | publish/subscribe and wildcard topics | | Error Handling | retriable vs terminal, the retry ladder, the DLQ | | Testing | unit, integration and end-to-end |

| Understand it | | |---|---| | Architecture | what a service creates in the broker, and why | | Message Flow | the wire format and the round trip | | Delivery Guarantees | acks, confirms, duplicates, the parked caller |

| Look it up | | |---|---| | Configuration | every environment variable and its default | | API reference | Context, MessageService, RunnableService, ServiceProxy | | Errors | every exported error class and when it is thrown | | Custom Types | extending the type system |

| Run it | | |---|---| | Troubleshooting | symptom, cause, fix | | Security | what actor does and does not prove | | Logging | levels, structured records, your own sink | | Queue Migration | changing settings on live queues | | Known Issues | current limitations | | Migration Guide | upgrading, including 1.x to 2.x |

See a real system in a minute

git clone https://github.com/ArielLaub/protobus && cd protobus && npm install
npm run docker:up
bash scripts/run-combat-sample.sh

Six services fight a battle royale over the bus — RPC, published events and graceful shutdown in one run — and the script asserts exactly one player survived. The source is sample/combatGame.

Requirements

• Node.js 20+ (enforced by engines; CI runs 20, 22 and 24)
• RabbitMQ 3.8+

Development

npm test                          # unit suite
npm run test:integration          # integration suite (starts RabbitMQ via Docker)
node scripts/check-doc-snippets.js  # compile and run every example in the docs
bash scripts/run-combat-sample.sh   # end-to-end sample

License

MIT — Copyright (c) 2018 Remarkable Games Ltd. See LICENSE.