refactor(reactive-rpc): 💡 remove server part from RpcPersistentClient

Author: streamich

src/reactive-rpc/browser/createBinaryWsRpcClient.ts

@@ -1,8 +1,8 @@
-import {CborJsonValueCodec} from "../../json-pack/codecs/cbor";
-import {Writer} from "../../util/buffers/Writer";
-import {RpcPersistentClient, WebSocketChannel} from "../common";
-import {RpcCodec} from "../common/codec/RpcCodec";
-import {BinaryRpcMessageCodec} from "../common/codec/binary";
+import {CborJsonValueCodec} from '../../json-pack/codecs/cbor';
+import {Writer} from '../../util/buffers/Writer';
+import {RpcPersistentClient, WebSocketChannel} from '../common';
+import {RpcCodec} from '../common/codec/RpcCodec';
+import {BinaryRpcMessageCodec} from '../common/codec/binary';
 
 export const createBinaryWsRpcClient = (url: string) => {
   const writer = new Writer(1024 * 4);

src/reactive-rpc/browser/createJsonWsRpcClient.ts

@@ -1,8 +1,8 @@
-import {JsonJsonValueCodec} from "../../json-pack/codecs/json";
-import {Writer} from "../../util/buffers/Writer";
-import {RpcPersistentClient, WebSocketChannel} from "../common";
-import {RpcCodec} from "../common/codec/RpcCodec";
-import {CompactRpcMessageCodec} from "../common/codec/compact";
+import {JsonJsonValueCodec} from '../../json-pack/codecs/json';
+import {Writer} from '../../util/buffers/Writer';
+import {RpcPersistentClient, WebSocketChannel} from '../common';
+import {RpcCodec} from '../common/codec/RpcCodec';
+import {CompactRpcMessageCodec} from '../common/codec/compact';
 
 export const createJsonWsRpcClient = (url: string) => {
   const writer = new Writer(1024 * 4);

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

@@ -1,23 +1,15 @@
+import * as msg from '../messages';
 import {firstValueFrom, Observable, ReplaySubject, timer} from 'rxjs';
 import {filter, first, share, switchMap, takeUntil} from 'rxjs/operators';
-import * as msg from '../messages';
 import {StreamingRpcClient, StreamingRpcClientOptions} from './client/StreamingRpcClient';
-import {ApiRpcCaller} from './caller/ApiRpcCaller';
-import {RpcDuplex} from '../rpc/RpcDuplex';
-import {RpcMessageStreamProcessor, RpcMessageStreamProcessorOptions} from './RpcMessageStreamProcessor';
 import {PersistentChannel, PersistentChannelParams} from '../channel';
-import {RpcCodec} from '../codec/RpcCodec';
+import type {RpcCodec} from '../codec/RpcCodec';
 
-export interface RpcPersistentClientParams<Ctx = unknown> {
+export interface RpcPersistentClientParams {
   channel: PersistentChannelParams;
   codec: RpcCodec;
   client?: Omit<StreamingRpcClientOptions, 'send'>;
 
-  /**
-   * @todo Remove this option. Remove server from here.
-   */
-  server?: Omit<RpcMessageStreamProcessorOptions<Ctx>, 'send'>;
-
   /**
    * Number of milliseconds to periodically send keep-alive ".ping" notification
    * messages. If not specified, will default to 15,000 (15 seconds). If 0, will
@@ -35,59 +27,44 @@ export interface RpcPersistentClientParams<Ctx = unknown> {
 /**
  * RPC client which automatically reconnects if disconnected.
  */
-export class RpcPersistentClient<Ctx = unknown> {
+export class RpcPersistentClient {
   public channel: PersistentChannel;
-  public rpc?: RpcDuplex<Ctx>;
-  public readonly rpc$ = new ReplaySubject<RpcDuplex<Ctx>>(1);
+  public rpc?: StreamingRpcClient;
+  public readonly rpc$ = new ReplaySubject<StreamingRpcClient>(1);
 
-  constructor(params: RpcPersistentClientParams<Ctx>) {
+  constructor(params: RpcPersistentClientParams) {
     const ping = params.ping ?? 15000;
     const codec = params.codec;
     const textEncoder = new TextEncoder();
     this.channel = new PersistentChannel(params.channel);
     this.channel.open$.pipe(filter((open) => open)).subscribe(() => {
       const close$ = this.channel.open$.pipe(filter((open) => !open));
-
-      const duplex = new RpcDuplex<Ctx>({
-        client: new StreamingRpcClient({
-          ...(params.client || {}),
-          send: (messages: msg.ReactiveRpcClientMessage[]): void => {
-            const encoded = codec.encode(messages, codec.req);
-            this.channel.send$(encoded).subscribe();
-          },
-        }),
-        server: new RpcMessageStreamProcessor<Ctx>({
-          ...(params.server || {
-            caller: new ApiRpcCaller({
-              api: {},
-            }),
-            onNotification: () => {},
-          }),
-          send: (messages: (msg.ReactiveRpcServerMessage | msg.NotificationMessage)[]): void => {
-            const encoded = codec.encode(messages, codec.req);
-            this.channel.send$(encoded).subscribe();
-          },
-        }),
+      const client = new StreamingRpcClient({
+        ...(params.client || {}),
+        send: (messages: msg.ReactiveRpcClientMessage[]): void => {
+          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, codec.res);
-        duplex.onMessages((messages instanceof Array ? messages : [messages]) as msg.ReactiveRpcMessage[], {} as Ctx);
+        client.onMessages((messages instanceof Array ? messages : [messages]) as msg.ReactiveRpcServerMessage[]);
       });
 
       // Send ping notifications to keep the connection alive.
       if (ping) {
         timer(ping, ping)
           .pipe(takeUntil(close$))
           .subscribe(() => {
-            duplex.notify(params.pingMethod || '.ping', undefined);
+            client.notify(params.pingMethod || '.ping', undefined);
           });
       }
 
       if (this.rpc) this.rpc.disconnect();
-      this.rpc = duplex;
-      this.rpc$.next(duplex);
+      this.rpc = client;
+      this.rpc$.next(client);
     });
   }