RxJS wrapper for gRPC and TypeScript

A package that wraps the callback and event based @grpc/grpc-js and grpc-web with Promises and RxJS Observables. Both clients as well as servers may be used reactively. Works well in conjunction with grpc-tools, grpc_tools_node_protoc_ts and protoc-gen-grpc-web.

Before

class ExampleService implements IExampleServer {
  incrementStream(call: grpc.ServerDuplexStream<OneNumber, OneNumber>): void {
    call.on('data', (number) => {
      call.write(new OneNumber().setA(number.getA() + 1));
    });
    call.on('end', () => call.end());
  }
}

After

defineService<IExampleServer>(ExampleService, {
  incrementStream(request: Observable<OneNumber>): Observable<OneNumber> {
    return request.pipe(map((number) => new OneNumber().setA(number.getA() + 1)));
  },
});

Table of Contents

• Installation
  • yarn
  • npm
• Usage
  • Node.js server implementation
    • Reactify an entire service
    • Reactify individual methods
    • Throwing errors
  • Node.js client
  • Web client

Installation

reactive-grpc requires rxjs, as well as either @grpc/grpc-js (for Node.js server applications) or grpc-web (for web frontends) to be added to your project.

yarn

yarn add reactive-grpc rxjs @grpc/grpc-js # or grpc-web

npm

npm install --save reactive-grpc rxjs @grpc/grpc-js # or grpc-web

Usage

Node.js server implementation

You can either "reactify" individual service methods or entire services. The latter is recommended, due to better type inference.

Reactify an entire service (recommended)

Use the defineService function to convert an object implementing the reactive service interface into a regular gRPC service. The reactive type is created automatically from the regular interface generated by grpc-tools.

import { interval, Observable } from 'rxjs';
import { map, reduce } from 'rxjs/operators';

import { defineService } from 'reactive-grpc/node';

// Generated by grpc-tools and grpc_tools_node_protoc_ts
import { OneNumber, TwoNumbers, Empty } from '../generated/service_pb';
import { ExampleService, IExampleServer } from '../generated/service_grpc_pb';

/** Reactive server of the example service. */
export default defineService<IExampleServer>(ExampleService, {
  async addTwoNumbers(request: TwoNumbers): Promise<OneNumber> {
    return new OneNumber().setA(request.getA() + request.getB());
  },
  addStreamOfNumbers(request: Observable<OneNumber>): Promise<OneNumber> {
    return request
      .pipe(
        reduce((acc, value) => acc + value.getA(), 0),
        map((value) => new OneNumber().setA(value)),
      )
      .toPromise();
  },
  getFibonacciSequence(request: Empty): Observable<OneNumber> {
    let a = 0;
    let b = 1;
    return interval(100).pipe(
      map(() => {
        const next = a + b;
        a = b;
        b = next;
        return new OneNumber().setA(a);
      })
    );
  },
  runningAverage(request: Observable<OneNumber>): Observable<OneNumber> {
    let average = 0;
    return request.pipe(
      map((value, index) => {
        average = (value.getA() + index * average) / (index + 1);
        return new OneNumber().setA(average);
      })
    );
  },
});

If you require advanced functionality from the standard gRPC call objects, such as reading the metadata or watching the cancellation status, you may simply add the usual argument:

defineService<IExampleServer>(ExampleService, {
  runningAverage(
    request: Observable<OneNumber>,
    call: grpc.ServerDuplexStream<OneNumber, OneNumber>,
  ): Observable<OneNumber> {
    ...
  },
}

Methods with Promise<ResponseType> return types can also return objects instead to include trailer metadata and flags, which would otherwise be provided when calling the callback function:

defineService<IExampleServer>(ExampleService, {
  async addTwoNumbers(request: TwoNumbers): Promise<OneNumber> {
    return {
      value: new OneNumber().setA(request.getA() + request.getB()),
      trailer: ...,
      flags: ...,
    };
  },
}

Reactify individual methods

Use defineUnaryMethod, defineRequestStreamMethod, defineResponseStreamMethod and defineBidirectionalStreamMethod to wrap your reactive function definitions. You can then assign the returned function to, for example, a member function of a non-reactive gRPC service:

import { interval, Observable } from 'rxjs';
import { map, reduce } from 'rxjs/operators';

import {
  defineUnaryMethod,
  defineRequestStreamMethod,
  defineResponseStreamMethod,
  defineBidirectionalStreamMethod,
} from 'reactive-grpc/node';

// Generated by grpc-tools and grpc_tools_node_protoc_ts
import { OneNumber, TwoNumbers, Empty } from '../generated/service_pb';
import { IExampleServer } from '../generated/service_grpc_pb';

/** Server of the example service that wraps each method individually. */
export default class ExampleServer implements IExampleServer {
  addStreamOfNumbers = defineRequestStreamMethod(function (request: Observable<OneNumber>): Promise<OneNumber> {
    return request
      .pipe(
        reduce((acc, value) => acc + value.getA(), 0),
        map((value) => new OneNumber().setA(value)),
      )
      .toPromise();
  });
}

Throwing errors

gRPC errors can be emitted by throwing instances of the RpcError class to reject the returned Promise:

defineService<IExampleServer>(ExampleService, {
  async addTwoNumbers(request: TwoNumbers): Promise<OneNumber> {
    throw new RpcError(RpcError.StatusCode.UNIMPLEMENTED, 'This method is not yet implemented!');
  },
}

For streaming responses, the error should be given to the the returned Observable:

import { Observable, throwError } from 'rxjs';

defineService<IExampleServer>(ExampleService, {
  getFibonacciSequence(request: Empty): Observable<OneNumber> {
    return throwError(new RpcError(RpcError.StatusCode.UNAUTHENTICATED, 'Client is not authenticated.'));
  },
}

Node.js client

To create a reactive gRPC client for use with Node.js, you must first create a regular client instance, and then apply reactive-grpc's reactifyNodeClient function:

import * as grpc from '@grpc/grpc-js';
import { reactifyNodeClient, RpcError } from 'reactive-grpc/node';

// Generated by grpc-tools and grpc_tools_node_protoc_ts
import { ExampleClient, ExampleService } from '../generated/service_grpc_pb';

const client = new ExampleClient(port, grpc.credentials.createInsecure());
const reactiveClient = reactifyNodeClient(ExampleService, client);

The new client then creates an API with Promises and Observables:

try {
  const inputStream = from([1, 2, 3, 4]).pipe(map((num) => new OneNumber().setA(num)));
  const sum = await reactiveClient.addNumbers(inputStream);
  console.log(sum.getA());
} catch (err) {
  if (err instanceof RpcError) {
    // Handle gRPC errors here.
  }
}

const observable = reactiveClient.getFibonacciNumbers(new OneNumber().setA(20));
observable.subscribe(
  (value) => console.log(value.getA()),
  (err) => console.log("Oh no!"), // Handle gRPC errors here.
  () => console.log("Done!"),
);

Optionally, you can also supply Metadata and CallOptions as the second and third parameters of the calls. To retrieve the standard gRPC call object, use .call on the returned Promise or Observable:

console.log(observable.call);

Note that unsubscribing from a server side stream will automatically cancel the request:

const observable = reactiveClient.getFibonacciNumbers(new OneNumber().setA(20));
// Will produce a gRPC CANCELLED error after receiving 5 values.
// The error will be ignored by the client.
observable.pipe(take(5)).subscribe(
  (value) => console.log(value.getA()),
  (err) => console.log('Oh no!'),
  () => console.log('Done!'),
);

Web client

To create a reactive gRPC client for use in a browser (e.g. with webpack), you must first create a regular client instance, and then apply reactive-grpc's reactifyWebClient function. Make sure to use the generated client which uses callbacks, not promises, for reactification:

import * as grpc from 'grpc-web';
import { reactifyWebClient } from 'reactive-grpc/web';

// Generated by protoc-gen-grpc-web
import { ExampleClient } from '../generated/service_grpc_pb';

const client = new ExampleClient(port);
const reactiveClient = reactifyWebClient(client);

The new client then creates an API with Promises and Observables:

try {
  const request = new TwoNumbers().setA(5).setB(6);
  const sum = await reactiveClient.addTwoNumbers(request);
  console.log(sum.getA());
} catch (err) {
  if (err instanceof RpcError) {
    // Handle gRPC errors here.
  }
}

const observable = reactiveClient.getFibonacciNumbers(new OneNumber().setA(20));
observable.subscribe(
  (value) => console.log(value.getA()),
  (err) => console.log('Oh no!'), // Handle gRPC errors here.
  () => console.log('Done!'),
);

Optionally, you can also supply Metadata as the second parameter of the calls. To retrieve the standard gRPC call object, use .call on the returned Promise or Observable:

console.log(observable.call);

Note that unsubscribing from a server side stream will automatically cancel the request:

const observable = reactiveClient.getFibonacciNumbers(new OneNumber().setA(20));
// Will produce a gRPC CANCELLED error after receiving 5 values.
// The error will be ignored by the client.
observable.pipe(take(5)).subscribe(
  (value) => console.log(value.getA()),
  (err) => console.log('Oh no!'),
  () => console.log('Done!'),
);