Add `WakuRelay.deleteObserver` to allow removal of observers

Useful when a React component add observers when mounting and needs to
delete it when unmounting.

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## Added
+- `WakuRelay.deleteObserver` to allow removal of observers, useful when a React component add observers when mounting and needs to delete it when unmounting. 
+
 ## [0.7.0] - 2021-06-15
 
 ### Changed

src/lib/waku_relay/index.spec.ts

@@ -140,6 +140,30 @@ describe('Waku Relay', () => {
       expect(allMessages[1].version).to.eq(barMessage.version);
       expect(allMessages[1].payloadAsUtf8).to.eq(barMessageText);
     });
+
+    it('Delete observer', async function () {
+      this.timeout(10000);
+
+      const messageText =
+        'Published on content topic with added then deleted observer';
+      const message = WakuMessage.fromUtf8String(
+        messageText,
+        'added-then-deleted-observer'
+      );
+
+      // The promise **fails** if we receive a message on this observer.
+      const receivedMsgPromise: Promise<WakuMessage> = new Promise(
+        (resolve, reject) => {
+          waku2.relay.addObserver(reject, ['added-then-deleted-observer']);
+          waku2.relay.deleteObserver(reject, ['added-then-deleted-observer']);
+          setTimeout(resolve, 500);
+        }
+      );
+      await waku1.relay.send(message);
+
+      await receivedMsgPromise;
+      // If it does not throw then we are good.
+    });
   });
 
   describe('Custom pubsub topic', () => {

src/lib/waku_relay/index.ts

@@ -144,6 +144,28 @@ export class WakuRelay extends Gossipsub implements Pubsub {
     }
   }
 
+  /**
+   * Remove an observer of new messages received via waku relay.
+   * Useful to ensure the same observer is not registered several time
+   * (e.g when loading React components)
+   */
+  deleteObserver(
+    callback: (message: WakuMessage) => void,
+    contentTopics: string[] = []
+  ): void {
+    if (contentTopics.length === 0) {
+      if (this.observers['']) {
+        this.observers[''].delete(callback);
+      }
+    } else {
+      contentTopics.forEach((contentTopic) => {
+        if (this.observers[contentTopic]) {
+          this.observers[contentTopic].delete(callback);
+        }
+      });
+    }
+  }
+
   /**
    * Return the relay peers we are connected to and we would publish a message to
    */