feat: migrate to AsyncIterable and remove Observable

Remove Observable<T> completely in favor of AsyncIterable<T>.

Uses the useAsyncIterable flag in ts-proto.

stephenh/ts-proto#605

Signed-off-by: Christian Stewart <[email protected]>

Author: paralin

.github/workflows/tests.yml

@@ -26,6 +26,7 @@ jobs:
         uses: actions/setup-node@v3
         with:
           node-version: ${{ matrix.node }}
+          cache: 'yarn'
       - name: Yarn install
         run: yarn install
       - name: Depcheck
@@ -34,7 +35,28 @@ jobs:
         run: yarn run build
       - name: Lint Javascript
         run: yarn run lint:js
+      - name: Test Js
+        run: yarn test:js
+      # Cache go build cache, used to speedup go test
+      # https://markphelps.me/posts/speed-up-your-go-builds-with-actions-cache/
+      - id: go-cache-paths
+        run: |
+          echo "::set-output name=go-build::$(go env GOCACHE)"
+          echo "::set-output name=go-mod::$(go env GOMODCACHE)"
+      - name: Go Build Cache
+        uses: actions/cache@v3
+        with:
+          path: ${{ steps.go-cache-paths.outputs.go-build }}
+          key: ${{ runner.os }}-go-build-${{ hashFiles('**/go.sum') }}
+      # Cache go mod cache, used to speedup builds
+      - name: Go Mod Cache
+        uses: actions/cache@v3
+        with:
+          path: ${{ steps.go-cache-paths.outputs.go-mod }}
+          key: ${{ runner.os }}-go-mod-${{ hashFiles('**/go.sum') }}
+      - name: Lint Go
+        run: yarn run lint:go
       - name: Test Go
         run: make test
       - name: Test integration of Go and TypeScript
-        run: make integration
+        run: yarn integration

Makefile

@@ -109,6 +109,7 @@ gents: $(PROTOWRAP) node_modules
 		--ts_proto_opt=forceLong=long \
 		--ts_proto_opt=oneof=unions \
 		--ts_proto_opt=outputServices=default,outputServices=generic-definitions \
+		--ts_proto_opt=useAsyncIterable=true \
 		--proto_path $$(pwd)/vendor \
 		--print_structure \
 		--only_specified_files \

README.md

@@ -19,19 +19,17 @@ Can use any Stream multiplexer: defaults to [libp2p-mplex] over a WebSocket.
 
 # Usage
 
-Starting with the [protobuf-project] repository on the "starpc" branch.
+Start with the [protobuf-project] template repository on the "starpc" branch.
+
+[protobuf-project]: https://github.com/aperturerobotics/protobuf-project/tree/starpc
 
 Use "git add" to add your new .proto files, then `yarn gen` to generate the
-TypeScript and Go code for them.
+TypeScript and Go code.
 
 # Examples
 
-See the [protobuf-project] template on the "starpc" branch.
-
 The demo/boilerplate project implements the Echo example below.
 
-[protobuf-project]: https://github.com/aperturerobotics/protobuf-project/tree/starpc
-
 This repository uses protowrap, see the [Makefile](./Makefile).
 
 ## Protobuf
@@ -118,6 +116,7 @@ This example demonstrates both the server and client:
 import { pipe } from 'it-pipe'
 import { createHandler, createMux, Server, Client, Conn } from 'srpc'
 import { EchoerDefinition, EchoerServer, runClientTest } from 'srpc/echo'
+import { pushable } from 'it-pushable'
 
 const mux = createMux()
 const echoer = new EchoerServer()
@@ -129,20 +128,31 @@ const serverConn = new Conn(server)
 pipe(clientConn, serverConn, clientConn)
 const client = new Client(clientConn.buildOpenStreamFunc())
 
+// call the unary rpc
 console.log('Calling Echo: unary call...')
 let result = await demoServiceClient.Echo({
   body: 'Hello world!',
 })
 console.log('success: output', result.body)
 
-const clientRequestStream = new Observable<EchoMsg>(subscriber => {
-  subscriber.next({body: 'Hello world from streaming request.'})
-  subscriber.complete()
-})
+// create a client -> server stream
+const clientRequestStream = pushable<EchoMsg>({objectMode: true})
+clientRequestStream.push({body: 'Hello world from streaming request.'})
+clientRequestStream.end()
 
+// call the client -> server streaming rpc
 console.log('Calling EchoClientStream: client -> server...')
 result = await demoServiceClient.EchoClientStream(clientRequestStream)
 console.log('success: output', result.body)
+
+// call the server -> client streaming rpc
+console.log('Calling EchoServerStream: server -> client...')
+const serverStream = demoServiceClient.EchoServerStream({
+  body: 'Hello world from server to client streaming request.',
+})
+for await (const msg of serverStream) {
+  console.log('server: output', msg.body)
+}
 ```
 
 ## WebSocket

doc/overview.md

@@ -4,34 +4,34 @@ For a service:
 
 ```protobuf
 syntax = "proto3";
-package web.demo;
+package simple;
 
 service DemoService {
-  rpc DemoEcho(DemoEchoMsg) returns (DemoEchoMsg) {}
+  rpc BidiStreaming(stream TestMessage) returns (stream TestMessage) {}
 }
 
-message DemoEchoMsg {
-  string msg = 1;
+message TestMessage {
+  string value = 1;
 }
 ```
 
 ts-proto generates a RPC interface like:
 
 ```typescript
 export interface DemoService {
-  DemoEcho(request: DemoEchoMsg): Promise<DemoEchoMsg>
+  BidiStreaming(request: AsyncIterable<TestMessage>): AsyncIterable<TestMsg>
 }
 
 export class DemoServiceClientImpl implements DemoService {
   private readonly rpc: Rpc
   constructor(rpc: Rpc) {
     this.rpc = rpc
-    this.DemoEcho = this.DemoEcho.bind(this)
+    this.BidiStreaming = this.BidiStreaming.bind(this)
   }
-  DemoEcho(request: DemoEchoMsg): Promise<DemoEchoMsg> {
-    const data = DemoEchoMsg.encode(request).finish()
-    const promise = this.rpc.request('web.demo.DemoService', 'DemoEcho', data)
-    return promise.then((data) => DemoEchoMsg.decode(new _m0.Reader(data)))
+  BidiStreaming(request: AsyncIterable<TestMessage>): AsyncIterable<TestMessage> {
+    const data = TestMessage.encodeTransform(request);
+    const result = this.rpc.bidirectionalStreamingRequest('simple.Test', 'BidiStreaming', data);
+    return TestMessage.decodeTransform(result);
   }
 }
 ```
@@ -48,18 +48,18 @@ interface Rpc {
   clientStreamingRequest(
     service: string,
     method: string,
-    data: Observable<Uint8Array>
+    data: AsyncIterable<Uint8Array>
   ): Promise<Uint8Array>
   serverStreamingRequest(
     service: string,
     method: string,
     data: Uint8Array
-  ): Observable<Uint8Array>
+  ): AsyncIterable<Uint8Array>
   bidirectionalStreamingRequest(
     service: string,
     method: string,
-    data: Observable<Uint8Array>
-  ): Observable<Uint8Array>
+    data: AsyncIterable<Uint8Array>
+  ): AsyncIterable<Uint8Array>
 }
 ```
 

e2e/e2e.ts

@@ -1,18 +1,20 @@
 import { pipe } from 'it-pipe'
 import { createHandler, createMux, Server, Client, Conn } from '../srpc'
 import { EchoerDefinition, EchoerServer, runClientTest } from '../echo'
+import { runRpcStreamTest } from '../echo/client-test'
 
 async function runRPC() {
   const mux = createMux()
-  const echoer = new EchoerServer()
-  mux.register(createHandler(EchoerDefinition, echoer))
   const server = new Server(mux)
+  const echoer = new EchoerServer(server)
+  mux.register(createHandler(EchoerDefinition, echoer))
 
   const clientConn = new Conn()
   const serverConn = new Conn(server)
   pipe(clientConn, serverConn, clientConn)
   const client = new Client(clientConn.buildOpenStreamFunc())
 
+  await runRpcStreamTest(client)
   await runClientTest(client)
 }
 
@@ -21,7 +23,6 @@ runRPC()
     console.log('finished successfully')
   })
   .catch((err) => {
-    console.log('failed')
     console.error(err)
     process.exit(1)
   })

e2e/e2e_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/aperturerobotics/starpc/echo"
+	"github.com/aperturerobotics/starpc/rpcstream"
 	"github.com/aperturerobotics/starpc/srpc"
 	"github.com/libp2p/go-libp2p/p2p/muxer/mplex"
 	mp "github.com/libp2p/go-mplex"
@@ -17,8 +18,8 @@ import (
 // RunE2E runs an end to end test with a callback.
 func RunE2E(t *testing.T, cb func(client echo.SRPCEchoerClient) error) {
 	// construct the server
-	echoServer := &echo.EchoServer{}
 	mux := srpc.NewMux()
+	echoServer := echo.NewEchoServer(mux)
 	if err := echo.SRPCRegisterEchoer(mux, echoServer); err != nil {
 		t.Fatal(err.Error())
 	}
@@ -179,3 +180,25 @@ func TestE2E_BidiStream(t *testing.T) {
 		return strm.Close()
 	})
 }
+
+func TestE2E_RpcStream(t *testing.T) {
+	ctx := context.Background()
+	RunE2E(t, func(client echo.SRPCEchoerClient) error {
+		openStreamFn := rpcstream.NewRpcStreamOpenStream(func(ctx context.Context) (rpcstream.RpcStream, error) {
+			return client.RpcStream(ctx)
+		}, "test")
+		proxiedClient := srpc.NewClient(openStreamFn)
+		proxiedSvc := echo.NewSRPCEchoerClient(proxiedClient)
+
+		// run a RPC proxied over another RPC
+		resp, err := proxiedSvc.Echo(ctx, &echo.EchoMsg{Body: "hello world"})
+		if err != nil {
+			return err
+		}
+		if resp.GetBody() != "hello world" {
+			return errors.Errorf("response body incorrect: %q", resp.GetBody())
+		}
+
+		return nil
+	})
+}

echo/client-test.ts

@@ -1,6 +1,7 @@
 import { Client } from '../srpc/index.js'
 import { EchoerClientImpl, EchoMsg } from './echo.pb.js'
-import { Observable } from 'rxjs'
+import { pushable } from 'it-pushable'
+import { buildRpcStreamOpenStream } from '../rpcstream/rpcstream.js'
 
 export async function runClientTest(client: Client) {
   const demoServiceClient = new EchoerClientImpl(client)
@@ -12,10 +13,9 @@ export async function runClientTest(client: Client) {
   console.log('success: output', result.body)
 
   // observable for client requests
-  const clientRequestStream = new Observable<EchoMsg>((subscriber) => {
-    subscriber.next({ body: 'Hello world from streaming request.' })
-    subscriber.complete()
-  })
+  const clientRequestStream = pushable<EchoMsg>({ objectMode: true })
+  clientRequestStream.push({ body: 'Hello world from streaming request.' })
+  clientRequestStream.end()
 
   console.log('Calling EchoClientStream: client -> server...')
   result = await demoServiceClient.EchoClientStream(clientRequestStream)
@@ -25,17 +25,22 @@ export async function runClientTest(client: Client) {
   const serverStream = demoServiceClient.EchoServerStream({
     body: 'Hello world from server to client streaming request.',
   })
-  await new Promise<void>((resolve, reject) => {
-    serverStream.subscribe({
-      next(result) {
-        console.log('server: output', result.body)
-      },
-      complete() {
-        resolve()
-      },
-      error(err: Error) {
-        reject(err)
-      },
-    })
-  })
+  for await (const msg of serverStream) {
+    console.log('server: output', msg.body)
+  }
+}
+
+// runRpcStreamTest tests a RPCStream.
+export async function runRpcStreamTest(client: Client) {
+  console.log('Calling RpcStream to open a RPC stream client...')
+  const service = new EchoerClientImpl(client)
+  const openStreamFn = buildRpcStreamOpenStream(
+    'test',
+    service.RpcStream.bind(service)
+  )
+  const proxiedClient = new Client(openStreamFn)
+  const proxiedService = new EchoerClientImpl(proxiedClient)
+  console.log('Calling Echo via RPC stream...')
+  const resp = await proxiedService.Echo({ body: 'hello world via proxy' })
+  console.log('rpc stream test: succeeded: response: ' + resp.body)
 }