feat(reactive-rpc): 🎸 support different request/response codecs in Websocket rpc

Author: streamich

src/__tests__/reactive-rpc/uws/ws/rx-rpc/RpcPersistentClient.spec.ts

@@ -21,32 +21,32 @@ if (process.env.TEST_E2E) {
     ['rpc.rx.compact.json', compact, json, json],
     ['rpc.rx.compact.cbor', compact, cbor, cbor],
     ['rpc.rx.compact.msgpack', compact, msgpack, msgpack],
-    // ['rpc.rx.compact.json-cbor', compact, json, cbor],
-    // ['rpc.rx.compact.json-msgpack', compact, json, msgpack],
-    // ['rpc.rx.compact.cbor-json', compact, cbor, json],
-    // ['rpc.rx.compact.cbor-msgpack', compact, cbor, msgpack],
-    // ['rpc.rx.compact.msgpack-json', compact, msgpack, json],
-    // ['rpc.rx.compact.msgpack-cbor', compact, msgpack, cbor],
+    ['rpc.rx.compact.json-cbor', compact, json, cbor],
+    ['rpc.rx.compact.json-msgpack', compact, json, msgpack],
+    ['rpc.rx.compact.cbor-json', compact, cbor, json],
+    ['rpc.rx.compact.cbor-msgpack', compact, cbor, msgpack],
+    ['rpc.rx.compact.msgpack-json', compact, msgpack, json],
+    ['rpc.rx.compact.msgpack-cbor', compact, msgpack, cbor],
 
     ['rpc.rx.binary.cbor', binary, cbor, cbor],
     ['rpc.rx.binary.msgpack', binary, msgpack, msgpack],
     ['rpc.rx.binary.json', binary, json, json],
-    // ['rpc.rx.binary.json-cbor', binary, json, cbor],
-    // ['rpc.rx.binary.json-msgpack', binary, json, msgpack],
-    // ['rpc.rx.binary.cbor-json', binary, cbor, json],
-    // ['rpc.rx.binary.cbor-msgpack', binary, cbor, msgpack],
-    // ['rpc.rx.binary.msgpack-json', binary, msgpack, json],
-    // ['rpc.rx.binary.msgpack-cbor', binary, msgpack, cbor],
+    ['rpc.rx.binary.json-cbor', binary, json, cbor],
+    ['rpc.rx.binary.json-msgpack', binary, json, msgpack],
+    ['rpc.rx.binary.cbor-json', binary, cbor, json],
+    ['rpc.rx.binary.cbor-msgpack', binary, cbor, msgpack],
+    ['rpc.rx.binary.msgpack-json', binary, msgpack, json],
+    ['rpc.rx.binary.msgpack-cbor', binary, msgpack, cbor],
 
     ['rpc.json2.verbose.json', jsonRpc2, json, json],
     ['rpc.json2.verbose.cbor', jsonRpc2, cbor, cbor],
     ['rpc.json2.verbose.msgpack', jsonRpc2, msgpack, msgpack],
-    // ['rpc.json2.verbose.json-cbor', jsonRpc2, json, cbor],
-    // ['rpc.json2.verbose.json-msgpack', jsonRpc2, json, msgpack],
-    // ['rpc.json2.verbose.cbor-json', jsonRpc2, cbor, json],
-    // ['rpc.json2.verbose.cbor-msgpack', jsonRpc2, cbor, msgpack],
-    // ['rpc.json2.verbose.msgpack-json', jsonRpc2, msgpack, json],
-    // ['rpc.json2.verbose.msgpack-cbor', jsonRpc2, msgpack, cbor],
+    ['rpc.json2.verbose.json-cbor', jsonRpc2, json, cbor],
+    ['rpc.json2.verbose.json-msgpack', jsonRpc2, json, msgpack],
+    ['rpc.json2.verbose.cbor-json', jsonRpc2, cbor, json],
+    ['rpc.json2.verbose.cbor-msgpack', jsonRpc2, cbor, msgpack],
+    ['rpc.json2.verbose.msgpack-json', jsonRpc2, msgpack, json],
+    ['rpc.json2.verbose.msgpack-cbor', jsonRpc2, msgpack, cbor],
   ];
 
   for (const [protocolSpecifier, msgCodec, reqCodec, resCodec] of cases) {

src/reactive-rpc/common/codec/RpcCodec.ts

@@ -11,17 +11,15 @@ export class RpcCodec {
     return specifier as RpcSpecifier;
   }
 
-  public encode(messages: ReactiveRpcMessage[]): Uint8Array {
-    const valueCodec = this.res;
+  public encode(messages: ReactiveRpcMessage[], valueCodec: JsonValueCodec): Uint8Array {
     const encoder = valueCodec.encoder;
     const writer = encoder.writer;
     writer.reset();
     this.msg.encodeBatch(valueCodec, messages);
     return writer.flush();
   }
 
-  public decode(data: Uint8Array): ReactiveRpcMessage[] {
-    const valueCodec = this.req;
+  public decode(data: Uint8Array, valueCodec: JsonValueCodec): ReactiveRpcMessage[] {
     const decoder = valueCodec.decoder;
     const reader = decoder.reader;
     reader.reset(data);

src/reactive-rpc/common/rpc/RpcPersistentClient.ts

@@ -12,6 +12,10 @@ export interface RpcPersistentClientParams<Ctx = unknown> {
   channel: PersistentChannelParams;
   codec: RpcCodec;
   client?: Omit<StreamingRpcClientOptions, 'send'>;
+
+  /**
+   * @todo Remove this option. Remove server from here.
+   */
   server?: Omit<RpcMessageStreamProcessorOptions<Ctx>, 'send'>;
 
   /**
@@ -28,6 +32,9 @@ export interface RpcPersistentClientParams<Ctx = unknown> {
   pingMethod?: string;
 }
 
+/**
+ * RPC client which automatically reconnects if disconnected.
+ */
 export class RpcPersistentClient<Ctx = unknown> {
   public channel: PersistentChannel;
   public rpc?: RpcDuplex<Ctx>;
@@ -45,7 +52,7 @@ export class RpcPersistentClient<Ctx = unknown> {
         client: new StreamingRpcClient({
           ...(params.client || {}),
           send: (messages: msg.ReactiveRpcClientMessage[]): void => {
-            const encoded = codec.encode(messages);
+            const encoded = codec.encode(messages, codec.req);
             this.channel.send$(encoded).subscribe();
           },
         }),
@@ -57,15 +64,15 @@ export class RpcPersistentClient<Ctx = unknown> {
             onNotification: () => {},
           }),
           send: (messages: (msg.ReactiveRpcServerMessage | msg.NotificationMessage)[]): void => {
-            const encoded = codec.encode(messages);
+            const encoded = codec.encode(messages, codec.req);
             this.channel.send$(encoded).subscribe();
           },
         }),
       });
 
       this.channel.message$.pipe(takeUntil(close$)).subscribe((data) => {
         const encoded = typeof data === 'string' ? textEncoder.encode(data) : new Uint8Array(data);
-        const messages = codec.decode(encoded);
+        const messages = codec.decode(encoded, codec.res);
         duplex.onMessages((messages instanceof Array ? messages : [messages]) as msg.ReactiveRpcMessage[], {} as Ctx);
       });
 

src/reactive-rpc/common/rpc/__tests__/RpcPersistentClient.spec.ts

@@ -36,7 +36,7 @@ test('on remote method execution, sends message over WebSocket only once', async
   await until(() => onSend.mock.calls.length === 1);
   expect(onSend).toHaveBeenCalledTimes(1);
   const message = onSend.mock.calls[0][0];
-  const decoded = codec.decode(message);
+  const decoded = codec.decode(message, codec.req);
   const messageDecoded = decoded[0];
   expect(messageDecoded).toBeInstanceOf(RequestCompleteMessage);
   expect(messageDecoded).toMatchObject(new RequestCompleteMessage(1, 'foo.bar', new Value({foo: 'bar'}, undefined)));