[WALL-25366] refactor WalletLinkRelay to remove RxJS dependencies (#988)

* remove Observable dependency from WalletLinkRelay: refactor publishEvent

* remove BehaviorSubject

* replace from with pure promise

* replace zip

* remove of, catchError, timeout

* introduce connection.setSessionConfigListner, and remove distinctUntilChanged, mergeMap

* resetAndReload timeout

* replace connection.incomingEvent$, remove filter

* simplify

* linked$

* connected

* connection.connectedListner

* cleanup

* remove subscriptions

* subscribeIncomingEvent

* Revert "subscribeIncomingEvent"

This reverts commit cd6e73b.

* make listner properties private

* Revert "Revert "subscribeIncomingEvent""

This reverts commit 5b75d5e.

* asdfads

* (ui instanceof WalletLinkRelayUI)

* tests

* test

* test distinctUntilChanged

* fix test

* typo

* nit: the _ param is not needed

* rpc method param format

apps/testapp/src/components/RpcMethods/index.ts

@@ -2,7 +2,7 @@ export type RpcMethod = {
   connected?: boolean;
   method: string;
   params: Array<{ key: string; required?: boolean }>;
-  format?: (data: Record<string, string>) => Record<string, string>;
+  format?: (data: Record<string, string>) => [Record<string, string>];
 };
 
 const ethRequestAccounts = {
@@ -18,9 +18,11 @@ const ethAccounts = {
 const switchEthereumChain = {
   method: 'wallet_switchEthereumChain',
   params: [{ key: 'chainId', required: true }],
-  format: (data: Record<string, string>) => ({
-    chainId: `0x${Number(data.chainId).toString(16)}`,
-  }),
+  format: (data: Record<string, string>) => [
+    {
+      chainId: `0x${Number(data.chainId).toString(16)}`,
+    },
+  ],
 };
 
 const addEthereumChain = {
@@ -36,18 +38,20 @@ const addEthereumChain = {
     { key: 'blockExplorerUrl' },
     { key: 'iconUrl' },
   ],
-  format: (data: Record<string, string>) => ({
-    chainId: `0x${Number(data.chainId).toString(16)}`,
-    chainName: data.chainName,
-    rpcUrls: [data.rpcUrl],
-    blockExplorerUrls: [data.blockExplorerUrls],
-    iconUrls: [data.iconUrl],
-    nativeCurrency: {
-      name: 'ETH',
-      symbol: 'ETH',
-      decimals: 18,
+  format: (data: Record<string, string>) => [
+    {
+      chainId: `0x${Number(data.chainId).toString(16)}`,
+      chainName: data.chainName,
+      rpcUrls: [data.rpcUrl],
+      blockExplorerUrls: [data.blockExplorerUrls],
+      iconUrls: [data.iconUrl],
+      nativeCurrency: {
+        name: 'ETH',
+        symbol: 'ETH',
+        decimals: 18,
+      },
     },
-  }),
+  ],
 };
 
 const personal_sign = {

packages/wallet-sdk/src/connection/WalletLinkConnection.ts

@@ -1,16 +1,7 @@
 // Copyright (c) 2018-2023 Coinbase, Inc. <https://www.coinbase.com/>
 // Licensed under the Apache License, version 2.0
 
-import {
-  BehaviorSubject,
-  iif,
-  Observable,
-  of,
-  ReplaySubject,
-  Subscription,
-  throwError,
-  timer,
-} from 'rxjs';
+import { BehaviorSubject, iif, Observable, of, Subscription, throwError, timer } from 'rxjs';
 import {
   catchError,
   delay,
@@ -66,7 +57,25 @@ export class WalletLinkConnection {
   private nextReqId = IntNumber(1);
   private connectedSubject = new BehaviorSubject(false);
   private linkedSubject = new BehaviorSubject(false);
-  private sessionConfigSubject = new ReplaySubject<SessionConfig>(1);
+
+  private sessionConfigListener?: (_: SessionConfig) => void;
+  setSessionConfigListener(listener: (_: SessionConfig) => void): void {
+    this.sessionConfigListener = listener;
+  }
+
+  private linkedListener?: (_: boolean) => void;
+  setLinkedListener(listener: (_: boolean) => void): void {
+    this.linkedListener = listener;
+  }
+
+  private connectedListener?: (_: boolean) => void;
+  setConnectedListener(listener: (_: boolean) => void): void {
+    this.connectedListener = listener;
+  }
+
+  setIncomingEventListener(listener: (_: ServerMessageEvent) => void): void {
+    this.subscribeIncomingEvent(listener);
+  }
 
   /**
    * Constructor
@@ -133,7 +142,10 @@ export class WalletLinkConnection {
           distinctUntilChanged(),
           catchError((_) => of(false))
         )
-        .subscribe((connected) => this.connectedSubject.next(connected))
+        .subscribe((connected) => {
+          this.connectedSubject.next(connected);
+          this.connectedListener?.(connected);
+        })
     );
 
     // send heartbeat every n seconds while connected
@@ -175,6 +187,7 @@ export class WalletLinkConnection {
             onlineGuests: msg.onlineGuests,
           });
           this.linkedSubject.next(msg.linked || msg.onlineGuests > 0);
+          this.linkedListener?.(msg.linked || msg.onlineGuests > 0);
         })
     );
 
@@ -189,7 +202,7 @@ export class WalletLinkConnection {
             sessionIdHash: Session.hash(sessionId),
             metadata_keys: msg && msg.metadata ? Object.keys(msg.metadata) : undefined,
           });
-          this.sessionConfigSubject.next({
+          this.sessionConfigListener?.({
             webhookId: msg.webhookId,
             webhookUrl: msg.webhookUrl,
             metadata: msg.metadata,
@@ -269,33 +282,31 @@ export class WalletLinkConnection {
   }
 
   /**
-   * Emit current session config if available, and subsequent updates
-   * @returns an Observable for the session config
+   * Subscribe to incoming Event messages
    */
-  public get sessionConfig$(): Observable<SessionConfig> {
-    return this.sessionConfigSubject.asObservable();
-  }
-
-  /**
-   * Emit incoming Event messages
-   * @returns an Observable for the messages
-   */
-  public get incomingEvent$(): Observable<ServerMessageEvent> {
-    return this.ws.incomingJSONData$.pipe(
-      filter((m) => {
-        if (m.type !== 'Event') {
-          return false;
-        }
-        const sme = m as ServerMessageEvent;
-        return (
-          typeof sme.sessionId === 'string' &&
-          typeof sme.eventId === 'string' &&
-          typeof sme.event === 'string' &&
-          typeof sme.data === 'string'
-        );
-      }),
-      map((m) => m as ServerMessageEvent)
-    );
+  private incomingEventSub?: Subscription;
+  private subscribeIncomingEvent(listener: (_: ServerMessageEvent) => void): void {
+    this.incomingEventSub?.unsubscribe();
+
+    this.incomingEventSub = this.ws.incomingJSONData$
+      .pipe(
+        filter((m) => {
+          if (m.type !== 'Event') {
+            return false;
+          }
+          const sme = m as ServerMessageEvent;
+          return (
+            typeof sme.sessionId === 'string' &&
+            typeof sme.eventId === 'string' &&
+            typeof sme.event === 'string' &&
+            typeof sme.data === 'string'
+          );
+        }),
+        map((m) => m as ServerMessageEvent)
+      )
+      .subscribe((m) => {
+        listener(m);
+      });
   }
 
   /**
@@ -304,21 +315,21 @@ export class WalletLinkConnection {
    * @param value
    * @returns an Observable that completes when successful
    */
-  public setSessionMetadata(key: string, value: string | null): Observable<void> {
+  public setSessionMetadata(key: string, value: string | null): Promise<void> {
     const message = ClientMessageSetSessionConfig({
       id: IntNumber(this.nextReqId++),
       sessionId: this.sessionId,
       metadata: { [key]: value },
     });
 
-    return this.onceConnected$.pipe(
-      flatMap((_) => this.makeRequest<ServerMessageOK | ServerMessageFail>(message)),
-      map((res) => {
+    return this.onceConnected$
+      .toPromise()
+      .then(() => this.makeRequest<ServerMessageOK | ServerMessageFail>(message).toPromise())
+      .then((res) => {
         if (isServerMessageFail(res)) {
           throw new Error(res.error || 'failed to set session metadata');
         }
-      })
-    );
+      });
   }
 
   /**