feat: Complete all RPC kinds

__tests__/e2e.test.ts

@@ -16,6 +16,7 @@ import {
   OrderingServiceConstructor,
   STREAM_ERROR,
   SubscribableServiceConstructor,
+  UploadableServiceConstructor,
   TestServiceConstructor,
 } from './fixtures/services';
 import { UNCAUGHT_ERROR } from '../router/result';
@@ -135,6 +136,36 @@ describe.each(codecs)(
       });
     });
 
+    test('stream with init message', async () => {
+      const [clientTransport, serverTransport] = getTransports();
+      const serviceDefs = { test: TestServiceConstructor() };
+      const server = await createServer(serverTransport, serviceDefs);
+      const client = createClient<typeof server>(clientTransport);
+
+      const [input, output, close] = await client.test.echoWithPrefix.stream({
+        prefix: 'test',
+      });
+      input.push({ msg: 'abc', ignore: false });
+      input.push({ msg: 'def', ignore: true });
+      input.push({ msg: 'ghi', ignore: false });
+      input.end();
+
+      const result1 = await iterNext(output);
+      assert(result1.ok);
+      expect(result1.payload).toStrictEqual({ response: 'test abc' });
+
+      const result2 = await iterNext(output);
+      assert(result2.ok);
+      expect(result2.payload).toStrictEqual({ response: 'test ghi' });
+
+      close();
+      await testFinishesCleanly({
+        clientTransports: [clientTransport],
+        serverTransport,
+        server,
+      });
+    });
+
     test('fallible stream', async () => {
       const [clientTransport, serverTransport] = getTransports();
       const serviceDefs = { test: FallibleServiceConstructor() };
@@ -232,6 +263,52 @@ describe.each(codecs)(
       });
     });
 
+    test('upload', async () => {
+      const [clientTransport, serverTransport] = getTransports();
+
+      const serviceDefs = { uploadable: UploadableServiceConstructor() };
+      const server = await createServer(serverTransport, serviceDefs);
+      const client = createClient<typeof server>(clientTransport);
+
+      const [addStream, addResult] =
+        await client.uploadable.addMultiple.upload();
+      addStream.push({ n: 1 });
+      addStream.push({ n: 2 });
+      addStream.end();
+      const result = await addResult;
+      assert(result.ok);
+      expect(result.payload).toStrictEqual({ result: 3 });
+      await testFinishesCleanly({
+        clientTransports: [clientTransport],
+        serverTransport,
+        server,
+      });
+    });
+
+    test('upload with init message', async () => {
+      const [clientTransport, serverTransport] = getTransports();
+
+      const serviceDefs = { uploadable: UploadableServiceConstructor() };
+      const server = await createServer(serverTransport, serviceDefs);
+      const client = createClient<typeof server>(clientTransport);
+
+      const [addStream, addResult] =
+        await client.uploadable.addMultipleWithPrefix.upload({
+          prefix: 'test',
+        });
+      addStream.push({ n: 1 });
+      addStream.push({ n: 2 });
+      addStream.end();
+      const result = await addResult;
+      assert(result.ok);
+      expect(result.payload).toStrictEqual({ result: 'test 3' });
+      await testFinishesCleanly({
+        clientTransports: [clientTransport],
+        serverTransport,
+        server,
+      });
+    });
+
     test('message order is preserved in the face of disconnects', async () => {
       const [clientTransport, serverTransport] = getTransports();
       const serviceDefs = { test: OrderingServiceConstructor() };

__tests__/fixtures/services.ts

@@ -45,6 +45,23 @@ export const TestServiceConstructor = () =>
         }
       },
     })
+    .defineProcedure('echoWithPrefix', {
+      type: 'stream',
+      init: Type.Object({ prefix: Type.String() }),
+      input: EchoRequest,
+      output: EchoResponse,
+      errors: Type.Never(),
+      async handler(_ctx, init, msgStream, returnStream) {
+        for await (const msg of msgStream) {
+          const req = msg.payload;
+          if (!req.ignore) {
+            returnStream.push(
+              reply(msg, Ok({ response: `${init.payload.prefix} ${req.msg}` })),
+            );
+          }
+        }
+      },
+    })
     .finalize();
 
 export const OrderingServiceConstructor = () =>
@@ -187,3 +204,44 @@ export const SubscribableServiceConstructor = () =>
       },
     })
     .finalize();
+
+export const UploadableServiceConstructor = () =>
+  ServiceBuilder.create('uploadable')
+    .initialState({})
+    .defineProcedure('addMultiple', {
+      type: 'upload',
+      input: Type.Object({ n: Type.Number() }),
+      output: Type.Object({ result: Type.Number() }),
+      errors: Type.Never(),
+      async handler(_ctx, msgStream) {
+        let result = 0;
+        let lastMsg;
+        for await (const msg of msgStream) {
+          const { n } = msg.payload;
+          result += n;
+          lastMsg = msg;
+        }
+        return reply(lastMsg!, Ok({ result: result }));
+      },
+    })
+    .defineProcedure('addMultipleWithPrefix', {
+      type: 'upload',
+      init: Type.Object({ prefix: Type.String() }),
+      input: Type.Object({ n: Type.Number() }),
+      output: Type.Object({ result: Type.String() }),
+      errors: Type.Never(),
+      async handler(_ctx, init, msgStream) {
+        let result = 0;
+        let lastMsg;
+        for await (const msg of msgStream) {
+          const { n } = msg.payload;
+          result += n;
+          lastMsg = msg;
+        }
+        return reply(
+          lastMsg!,
+          Ok({ result: init.payload.prefix + ' ' + result }),
+        );
+      },
+    })
+    .finalize();

__tests__/handler.test.ts

@@ -1,7 +1,10 @@
 import {
   asClientRpc,
   asClientStream,
+  asClientStreamWithInitialization,
   asClientSubscription,
+  asClientUpload,
+  asClientUploadWithInitialization,
   iterNext,
 } from '../util/testHelpers';
 import { assert, describe, expect, test } from 'vitest';
@@ -10,6 +13,7 @@ import {
   FallibleServiceConstructor,
   STREAM_ERROR,
   SubscribableServiceConstructor,
+  UploadableServiceConstructor,
   TestServiceConstructor,
 } from './fixtures/services';
 import { UNCAUGHT_ERROR } from '../router/result';
@@ -73,6 +77,29 @@ describe('server-side test', () => {
     expect(output.readableLength).toBe(0);
   });
 
+  test('stream with initialization', async () => {
+    const [input, output] = asClientStreamWithInitialization(
+      initialState,
+      service.procedures.echoWithPrefix,
+      { prefix: 'test' },
+    );
+
+    input.push({ msg: 'abc', ignore: false });
+    input.push({ msg: 'def', ignore: true });
+    input.push({ msg: 'ghi', ignore: false });
+    input.end();
+
+    const result1 = await iterNext(output);
+    assert(result1 && result1.ok);
+    expect(result1.payload).toStrictEqual({ response: 'test abc' });
+
+    const result2 = await iterNext(output);
+    assert(result2 && result2.ok);
+    expect(result2.payload).toStrictEqual({ response: 'test ghi' });
+
+    expect(output.readableLength).toBe(0);
+  });
+
   test('fallible stream', async () => {
     const service = FallibleServiceConstructor();
     const [input, output] = asClientStream({}, service.procedures.echo);
@@ -118,4 +145,31 @@ describe('server-side test', () => {
     assert(streamResult2 && streamResult1.ok);
     expect(streamResult2.payload).toStrictEqual({ result: 3 });
   });
+
+  test('uploads', async () => {
+    const service = UploadableServiceConstructor();
+    const [input, result] = asClientUpload({}, service.procedures.addMultiple);
+
+    input.push({ n: 1 });
+    input.push({ n: 2 });
+    input.end();
+    expect(await result).toStrictEqual({ ok: true, payload: { result: 3 } });
+  });
+
+  test('uploads with initialization', async () => {
+    const service = UploadableServiceConstructor();
+    const [input, result] = asClientUploadWithInitialization(
+      {},
+      service.procedures.addMultipleWithPrefix,
+      { prefix: 'test' },
+    );
+
+    input.push({ n: 1 });
+    input.push({ n: 2 });
+    input.end();
+    expect(await result).toStrictEqual({
+      ok: true,
+      payload: { result: 'test 3' },
+    });
+  });
 });

__tests__/serialize.test.ts

@@ -51,6 +51,45 @@ describe('serialize service to jsonschema', () => {
           errors: { not: {} },
           type: 'stream',
         },
+        echoWithPrefix: {
+          errors: {
+            not: {},
+          },
+          init: {
+            properties: {
+              prefix: {
+                type: 'string',
+              },
+            },
+            required: ['prefix'],
+            type: 'object',
+          },
+          input: {
+            properties: {
+              end: {
+                type: 'boolean',
+              },
+              ignore: {
+                type: 'boolean',
+              },
+              msg: {
+                type: 'string',
+              },
+            },
+            required: ['msg', 'ignore'],
+            type: 'object',
+          },
+          output: {
+            properties: {
+              response: {
+                type: 'string',
+              },
+            },
+            required: ['response'],
+            type: 'object',
+          },
+          type: 'stream',
+        },
       },
     });
   });

package.json

@@ -2,7 +2,7 @@
   "name": "@replit/river",
   "sideEffects": false,
   "description": "It's like tRPC but... with JSON Schema Support, duplex streaming and support for service multiplexing. Transport agnostic!",
-  "version": "0.8.1",
+  "version": "0.9.0",
   "type": "module",
   "exports": {
     ".": "./dist/router/index.js",