@replit/river

river - Streaming Remote Procedure Calls

It's like tRPC/gRPC but with

• JSON Schema Support + run-time schema validation
• full-duplex streaming
• service multiplexing
• result types and error handling
• snappy DX (no code-generation)
• over any transport (WebSockets, stdio, Unix Domain Socket out of the box)

Installation

To use River, you must be on least Typescript 5 with "moduleResolution": "bundler".

npm i @replit/river @sinclair/typebox

# if you plan on using WebSocket for transport, also install
npm i ws isomorphic-ws

Writing Services

Concepts

• Router: a collection of services, namespaced by service name.
• Service: a collection of procedures with shared state.
• Procedure: a single procedure. A procedure declares its type, an input message type, an output message type, optionally an error type, and the associated handler. Valid types are:
  • rpc whose handler has a signature of Input -> Result<Output, Error>.
  • upload whose handler has a signature of AsyncIterableIterator<Input> -> Result<Output, Error>.
  • subscription whose handler has a signature of Input -> Pushable<Result<Output, Error>>.
  • stream whose handler has a signature of AsyncIterableIterator<Input> -> Pushable<Result<Output, Error>>.
• Transport: manages the lifecycle (creation/deletion) of connections and multiplexing read/writes from clients. Both the client and the server must be passed in a subclass of Transport to work.
• Codec: encodes messages between clients/servers before the transport sends it across the wire.

A basic router

First, we create a service using the ServiceBuilder

import { ServiceBuilder, Ok, buildServiceDefs } from '@replit/river';
import { Type } from '@sinclair/typebox';

export const ExampleServiceConstructor = () =>
  ServiceBuilder.create('example')
    .initialState({
      count: 0,
    })
    .defineProcedure('add', {
      type: 'rpc',
      input: Type.Object({ n: Type.Number() }),
      output: Type.Object({ result: Type.Number() }),
      errors: Type.Never(),
      async handler(ctx, { n }) {
        ctx.state.count += n;
        return Ok({ result: ctx.state.count });
      },
    })
    .finalize();

// expore a listing of all the services that we have
export const serviceDefs = buildServiceDefs([ExampleServiceConstructor()]);

Then, we create the server

import http from 'http';
import { WebSocketServer } from 'ws';
import { WebSocketServerTransport } from '@replit/river/transport/ws/server';
import { createServer } from '@replit/river';

// start websocket server on port 3000
const httpServer = http.createServer();
const port = 3000;
const wss = new WebSocketServer({ server: httpServer });
const transport = new WebSocketServerTransport(wss, 'SERVER');

export const server = createServer(transport, serviceDefs);
export type ServiceSurface = typeof server;

httpServer.listen(port);

In another file for the client (to create a separate entrypoint),

import WebSocket from 'isomorphic-ws';
import { WebSocketClientTransport } from '@replit/river/transport/ws/client';
import { createClient } from '@replit/river';
import type ServiceSurface from './server';

const websocketUrl = `ws://localhost:3000`;
const transport = new WebSocketClientTransport(
  async () => new WebSocket(websocketUrl),
  'my-client-id',
  'SERVER',
);

const client = createClient<ServiceSurface>(transport, 'SERVER');

// we get full type safety on `client`
// client.<service name>.<procedure name>.<procedure type>()
// e.g.
const result = await client.example.add.rpc({ n: 3 });
if (result.ok) {
  const msg = result.payload;
  console.log(msg.result); // 0 + 3 = 3
}

To add logging,

import { bindLogger, setLevel } from '@replit/river/logging';

bindLogger(console.log);
setLevel('info');

To listen for connection status changes,

transport.addEventListener('connectionStatus', (evt) => {
  if (evt.status === 'connect') {
    // do something
  } else if (evt.status === 'disconnect') {
    // do something else
  }
});

[!note] WebSocket connection behaviour WebSocket is an idle protocol. This means that when the underlying connection drops, the WebSocket may still think it is still connected (e.g. turning off the network via devtools). You can use window.addEventListener('online', ...); and window.addEventListener('offline', ...); to know if you need to recreate the transport.

Further examples

We've also provided an end-to-end testing environment using Next.js, and a simple backend connected with the WebSocket transport that you can play with on Replit.

You can find more service examples in the E2E test fixtures

Developing

• npm i -- install dependencies
• npm run check -- lint
• npm run format -- format
• npm run test -- run tests
• npm run publish -- cut a new release (should bump version in package.json first)