From 275b16641e620956a5f8ebbb3a8c4156149d489e Mon Sep 17 00:00:00 2001
From: Sasha <118575614+weboko@users.noreply.github.com>
Date: Wed, 15 Mar 2023 21:47:56 +0100
Subject: [PATCH] feat!: add custom events to Relay and make observers private
 (#1213)

---
 packages/core/src/lib/relay/index.ts | 29 +++++++++++++++++++++++++---
 packages/interfaces/src/relay.ts     | 14 ++++++++++++--
 2 files changed, 38 insertions(+), 5 deletions(-)

diff --git a/packages/core/src/lib/relay/index.ts b/packages/core/src/lib/relay/index.ts
index b588a79bb5..296282133b 100644
--- a/packages/core/src/lib/relay/index.ts
+++ b/packages/core/src/lib/relay/index.ts
@@ -6,6 +6,7 @@ import {
 } from "@chainsafe/libp2p-gossipsub";
 import type { PeerIdStr, TopicStr } from "@chainsafe/libp2p-gossipsub/types";
 import { SignaturePolicy } from "@chainsafe/libp2p-gossipsub/types";
+import { CustomEvent } from "@libp2p/interfaces/events";
 import type {
   Callback,
   IDecoder,
@@ -35,6 +36,10 @@ export type Observer<T extends IDecodedMessage> = {
 export type RelayCreateOptions = ProtocolCreateOptions & GossipsubOpts;
 export type ContentTopic = string;
 
+type BasicEventPayload = {
+  contentTopic: string;
+};
+
 /**
  * Implements the [Waku v2 Relay protocol](https://rfc.vac.dev/spec/11/).
  * Must be passed as a `pubsub` module to a `Libp2p` instance.
@@ -50,7 +55,7 @@ class Relay extends GossipSub implements IRelay {
    * observers called when receiving new message.
    * Observers under key `""` are always called.
    */
-  public observers: Map<ContentTopic, Set<unknown>>;
+  private observers: Map<ContentTopic, Set<unknown>>;
 
   constructor(
     components: GossipSubComponents,
@@ -111,12 +116,30 @@ class Relay extends GossipSub implements IRelay {
       decoder,
       callback,
     };
-    pushOrInitMapSet(this.observers, decoder.contentTopic, observer);
+    const contentTopic = decoder.contentTopic;
+
+    pushOrInitMapSet(this.observers, contentTopic, observer);
+
+    this.dispatchEvent(
+      new CustomEvent<BasicEventPayload>("observer:added", {
+        detail: {
+          contentTopic,
+        },
+      })
+    );
 
     return () => {
-      const observers = this.observers.get(decoder.contentTopic);
+      const observers = this.observers.get(contentTopic);
       if (observers) {
         observers.delete(observer);
+
+        this.dispatchEvent(
+          new CustomEvent<BasicEventPayload>("observer:removed", {
+            detail: {
+              contentTopic,
+            },
+          })
+        );
       }
     };
   }
diff --git a/packages/interfaces/src/relay.ts b/packages/interfaces/src/relay.ts
index f88e4ca676..d90dfb3a1f 100644
--- a/packages/interfaces/src/relay.ts
+++ b/packages/interfaces/src/relay.ts
@@ -1,4 +1,5 @@
-import type { GossipSub } from "@chainsafe/libp2p-gossipsub";
+import type { GossipSub, GossipsubEvents } from "@chainsafe/libp2p-gossipsub";
+import type { EventEmitter } from "@libp2p/interfaces/events";
 
 import type {
   IDecodedMessage,
@@ -8,7 +9,14 @@ import type {
 } from "./message.js";
 import type { Callback, SendResult } from "./protocols.js";
 
-export interface IRelay extends GossipSub {
+export interface RelayEvents {
+  "observer:added": CustomEvent;
+  "observer:removed": CustomEvent;
+}
+
+type IRelayEmitter = EventEmitter<RelayEvents & GossipsubEvents>;
+
+interface IRelayAPI extends GossipSub {
   send: (encoder: IEncoder, message: IMessage) => Promise<SendResult>;
   addObserver: <T extends IDecodedMessage>(
     decoder: IDecoder<T>,
@@ -16,3 +24,5 @@ export interface IRelay extends GossipSub {
   ) => () => void;
   getMeshPeers: () => string[];
 }
+
+export type IRelay = IRelayAPI & IRelayEmitter;