client should close on server-side stream close bit

replit · Dec 8, 2023 · cf4d598 · cf4d598
1 parent 7c8d121
commit cf4d598
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 4 deletions.
diff --git a/__tests__/e2e.test.ts b/__tests__/e2e.test.ts
@@ -91,6 +91,7 @@ describe.each(codecs)(
       input.push({ msg: 'abc', ignore: false });
       input.push({ msg: 'def', ignore: true });
       input.push({ msg: 'ghi', ignore: false });
+      input.push({ msg: 'end', ignore: false, end: true });
       input.end();
 
       const result1 = await iterNext(output);
@@ -101,6 +102,13 @@ describe.each(codecs)(
       assert(result2.ok);
       expect(result2.payload).toStrictEqual({ response: 'ghi' });
 
+      const result3 = await iterNext(output);
+      assert(result3.ok);
+      expect(result3.payload).toStrictEqual({ response: 'end' });
+
+      const result4 = await output.next();
+      assert(result4.done);
+
       close();
     });
 

diff --git a/__tests__/fixtures/services.ts b/__tests__/fixtures/services.ts
@@ -7,6 +7,7 @@ import { Observable } from './observable';
 export const EchoRequest = Type.Object({
   msg: Type.String(),
   ignore: Type.Boolean(),
+  end: Type.Optional(Type.Boolean()),
 });
 export const EchoResponse = Type.Object({ response: Type.String() });
 
@@ -37,6 +38,10 @@ export const TestServiceConstructor = () =>
           if (!req.ignore) {
             returnStream.push(reply(msg, Ok({ response: req.msg })));
           }
+
+          if (req.end) {
+            returnStream.end();
+          }
         }
       },
     })

diff --git a/__tests__/serialize.test.ts b/__tests__/serialize.test.ts
@@ -36,6 +36,7 @@ describe('serialize service to jsonschema', () => {
             properties: {
               msg: { type: 'string' },
               ignore: { type: 'boolean' },
+              end: { type: 'boolean' },
             },
             required: ['msg', 'ignore'],
             type: 'object',

diff --git a/router/client.ts b/router/client.ts
@@ -180,12 +180,10 @@ export const createClient = <Srv extends Server<Record<string, AnyService>>>(
 
       // transport -> output
       const listener = (msg: OpaqueTransportMessage) => {
-        if (belongsToSameStream(msg)) {
-          outputStream.push(msg.payload);
-        }
-
         if (isStreamClose(msg.controlFlags)) {
           outputStream.end();
+        } else if (belongsToSameStream(msg)) {
+          outputStream.push(msg.payload);
         }
       };