fix: changed client stream caller signature

Related #501

[ci skip]

Author: tegefaulkes

src/RPC/RPCClient.ts

@@ -75,7 +75,7 @@ class RPCClient<M extends ClientManifest> {
           case 'SERVER':
             return (params) => this.serverStreamCaller(method, params);
           case 'CLIENT':
-            return (f) => this.clientStreamCaller(method, f);
+            return () => this.clientStreamCaller(method);
           case 'DUPLEX':
             return (f) => this.duplexStreamCaller(method, f);
           case 'RAW':
@@ -142,8 +142,6 @@ class RPCClient<M extends ClientManifest> {
     return this.rawMethodsProxy as MapRawCallers<M>;
   }
 
-  // Convenience methods
-
   @ready(new rpcErrors.ErrorRpcDestroyed())
   public async unaryCaller<I extends JSONValue, O extends JSONValue>(
     method: string,
@@ -183,21 +181,22 @@ class RPCClient<M extends ClientManifest> {
   @ready(new rpcErrors.ErrorRpcDestroyed())
   public async clientStreamCaller<I extends JSONValue, O extends JSONValue>(
     method: string,
-    f: (output: Promise<O>) => AsyncIterable<I | undefined>,
-  ): Promise<void> {
-    const callerInterface = await this.rawClientStreamCaller<I, O>(method);
-    const writer = callerInterface.writable.getWriter();
-    let running = true;
-    for await (const value of f(callerInterface.output)) {
-      if (value === undefined) {
-        await writer.close();
-        running = false;
+  ): Promise<{
+    output: Promise<O>;
+    writable: WritableStream<I>;
+  }> {
+    const callerInterface = await this.rawDuplexStreamCaller<I, O>(method);
+    const reader = callerInterface.readable.getReader();
+    const output = reader.read().then(({ value, done }) => {
+      if (done) {
+        throw new rpcErrors.ErrorRpcRemoteError('Stream ended before response');
       }
-      // Write while running otherwise consume until ended
-      if (running) await writer.write(value);
-    }
-    // If ended before finish running then close writer
-    if (running) await writer.close();
+      return value;
+    });
+    return {
+      output,
+      writable: callerInterface.writable,
+    };
   }
 
   @ready(new rpcErrors.ErrorRpcDestroyed())
@@ -263,29 +262,6 @@ class RPCClient<M extends ClientManifest> {
     tempWriter.releaseLock();
     return streamPair;
   }
-
-  protected async rawClientStreamCaller<
-    I extends JSONValue,
-    O extends JSONValue,
-  >(
-    method: string,
-  ): Promise<{
-    output: Promise<O>;
-    writable: WritableStream<I>;
-  }> {
-    const callerInterface = await this.rawDuplexStreamCaller<I, O>(method);
-    const reader = callerInterface.readable.getReader();
-    const output = reader.read().then(({ value, done }) => {
-      if (done) {
-        throw new rpcErrors.ErrorRpcRemoteError('Stream ended before response');
-      }
-      return value;
-    });
-    return {
-      output,
-      writable: callerInterface.writable,
-    };
-  }
 }
 
 export default RPCClient;

src/RPC/types.ts

@@ -161,7 +161,7 @@ type ServerCallerImplementation<
 type ClientCallerImplementation<
   I extends JSONValue = JSONValue,
   O extends JSONValue = JSONValue,
-> = (f: (output: Promise<O>) => AsyncIterable<I | undefined>) => Promise<void>;
+> = () => Promise<{ output: Promise<O>; writable: WritableStream<I> }>;
 
 type DuplexCallerImplementation<
   I extends JSONValue = JSONValue,

tests/RPC/RPC.test.ts

@@ -208,17 +208,14 @@ describe('RPC', () => {
         logger,
       });
 
-      await rpcClient.methods.testMethod(async function* (output) {
-        for (const value of values) {
-          yield value;
-        }
-        // Ending writes
-        yield undefined;
-        // Checking output
-        const expectedResult = values.reduce((p, c) => p + c);
-        await expect(output).resolves.toEqual(expectedResult);
-      });
-
+      const { output, writable } = await rpcClient.methods.testMethod();
+      const writer = writable.getWriter();
+      for (const value of values) {
+        await writer.write(value);
+      }
+      await writer.close();
+      const expectedResult = values.reduce((p, c) => p + c);
+      await expect(output).resolves.toEqual(expectedResult);
       await rpcServer.destroy();
       await rpcClient.destroy();
     },

tests/RPC/RPCClient.test.ts

@@ -175,16 +175,16 @@ describe(`${RPCClient.name}`, () => {
         streamPairCreateCallback: async () => streamPair,
         logger,
       });
-      await rpcClient.clientStreamCaller<JSONValue, JSONValue>(
-        methodName,
-        async function* (output) {
-          for (const param of params) {
-            yield param;
-          }
-          yield undefined;
-          expect(await output).toStrictEqual(message.result);
-        },
-      );
+      const { output, writable } = await rpcClient.clientStreamCaller<
+        JSONValue,
+        JSONValue
+      >(methodName);
+      const writer = writable.getWriter();
+      for (const param of params) {
+        await writer.write(param);
+      }
+      await writer.close();
+      expect(await output).toStrictEqual(message.result);
       const expectedOutput = params.map((v) =>
         JSON.stringify({
           method: methodName,
@@ -518,13 +518,13 @@ describe(`${RPCClient.name}`, () => {
         streamPairCreateCallback: async () => streamPair,
         logger,
       });
-      await rpcClient.methods.client(async function* (output) {
-        for (const param of params) {
-          yield param;
-        }
-        yield undefined;
-        expect(await output).toStrictEqual(message.result);
-      });
+      const { output, writable } = await rpcClient.methods.client();
+      const writer = writable.getWriter();
+      for (const param of params) {
+        await writer.write(param);
+      }
+      expect(await output).toStrictEqual(message.result);
+      await writer.close();
       const expectedOutput = params.map((v) =>
         JSON.stringify({
           method: 'client',