Add method to unsubscribe from daemon events

2 files changed, 43 insertions, 7 deletions

diff --git a/gui/src/main/daemon-rpc.ts b/gui/src/main/daemon-rpc.ts
index 4e349aae36..481833e0ac 100644
--- a/gui/src/main/daemon-rpc.ts
+++ b/gui/src/main/daemon-rpc.ts
@@ -303,6 +303,10 @@ export class ConnectionObserver {
 }
 
 export class SubscriptionListener<T> {
+  // Only meant to be used by DaemonRpc
+  // @internal
+  public subscriptionId?: string | number;
+
   constructor(
     private eventHandler: (payload: T) => void,
     private errorHandler: (error: Error) => void,
@@ -493,17 +497,31 @@ export class DaemonRpc {
     }
   }
 
-  public subscribeDaemonEventListener(listener: SubscriptionListener<DaemonEvent>): Promise<void> {
-    return this.transport.subscribe('daemon_event', (payload) => {
+  public async subscribeDaemonEventListener(
+    listener: SubscriptionListener<DaemonEvent>,
+  ): Promise<void> {
+    const subscriptionId = await this.transport.subscribe('daemon_event', (payload) => {
+      let daemonEvent: DaemonEvent;
+
       try {
-        const daemonEvent = camelCaseObjectKeys(
-          validate(daemonEventSchema, payload),
-        ) as DaemonEvent;
-        listener.onEvent(daemonEvent);
+        daemonEvent = camelCaseObjectKeys(validate(daemonEventSchema, payload)) as DaemonEvent;
       } catch (error) {
         listener.onError(new ResponseParseError('Invalid payload from daemon_event', error));
+        return;
       }
+
+      listener.onEvent(daemonEvent);
     });
+
+    listener.subscriptionId = subscriptionId;
+  }
+
+  public async unsubscribeDaemonEventListener(
+    listener: SubscriptionListener<DaemonEvent>,
+  ): Promise<void> {
+    if (listener.subscriptionId) {
+      return this.transport.unsubscribe('daemon_event', listener.subscriptionId);
+    }
   }
 
   public async getAccountHistory(): Promise<AccountToken[]> {
diff --git a/gui/src/main/jsonrpc-client.ts b/gui/src/main/jsonrpc-client.ts
index f3ac7e78e1..357d085a14 100644
--- a/gui/src/main/jsonrpc-client.ts
+++ b/gui/src/main/jsonrpc-client.ts
@@ -163,11 +163,12 @@ export default class JsonRpcClient<T> extends EventEmitter {
     }
   }
 
-  public async subscribe(event: string, listener: (value: any) => void): Promise<void> {
+  public async subscribe(event: string, listener: (value: any) => void): Promise<string | number> {
     log.silly(`Adding a listener for ${event}`);
 
     try {
       const subscriptionId = await this.send(`${event}_subscribe`);
+
       if (typeof subscriptionId === 'string' || typeof subscriptionId === 'number') {
         this.subscriptions.set(subscriptionId, listener);
       } else {
@@ -176,12 +177,29 @@ export default class JsonRpcClient<T> extends EventEmitter {
           subscriptionId,
         );
       }
+
+      return subscriptionId;
     } catch (e) {
       log.error(`Failed adding listener to ${event}: ${e.message}`);
       throw e;
     }
   }
 
+  public async unsubscribe(event: string, subscriptionId: string | number): Promise<void> {
+    log.silly(`Removing a listener for ${event}`);
+
+    try {
+      if (this.subscriptions.has(subscriptionId)) {
+        await this.send(`${event}_unsubscribe`, [subscriptionId]);
+      }
+    } catch (e) {
+      log.error(`Failed removing listener to ${event}: ${e.message}`);
+      throw e;
+    } finally {
+      this.subscriptions.delete(subscriptionId);
+    }
+  }
+
   public send(action: string, data?: any, timeout: number = DEFAULT_TIMEOUT_MILLIS): Promise<any> {
     return new Promise((resolve, reject) => {
       const transport = this.transport;