feat: track node connection state

Author: adklempner

packages/core/src/lib/connection_manager.ts

@@ -40,6 +40,25 @@ export class ConnectionManager
 
   private currentActiveParallelDialCount = 0;
   private pendingPeerDialQueue: Array<PeerId> = [];
+  private online: boolean = false;
+
+  public isConnected(): boolean {
+    return this.online;
+  }
+
+  private toggleOnline(): void {
+    if (!this.online) {
+      this.online = true;
+      this.dispatchEvent(new CustomEvent(EPeersByDiscoveryEvents.NODE_ONLINE));
+    }
+  }
+
+  private toggleOffline(): void {
+    if (this.libp2p.getPeers().length == 0) {
+      this.online = false;
+      this.dispatchEvent(new CustomEvent(EPeersByDiscoveryEvents.NODE_OFFLINE));
+    }
+  }
 
   public static create(
     peerId: string,
@@ -393,11 +412,13 @@ export class ConnectionManager
             )
           );
         }
+        this.toggleOnline();
       })();
     },
     "peer:disconnect": () => {
       return (evt: CustomEvent<PeerId>): void => {
         this.keepAliveManager.stop(evt.detail);
+        this.toggleOffline();
       };
     }
   };
@@ -427,7 +448,7 @@ export class ConnectionManager
       log.warn(
         `Discovered peer ${peerId.toString()} with ShardInfo ${shardInfo} is not part of any of the configured pubsub topics (${
           this.configuredPubSubTopics
-        }). 
+        }).
             Not dialing.`
       );
       return false;

packages/core/src/lib/waku.ts

@@ -178,6 +178,10 @@ export class WakuNode implements Waku {
     return this.libp2p.isStarted();
   }
 
+  isConnected(): boolean {
+    return this.connectionManager.isConnected();
+  }
+
   /**
    * Return the local multiaddr with peer id on which libp2p is listening.
    *

packages/interfaces/src/connection_manager.ts

@@ -28,14 +28,18 @@ export enum EPeersByDiscoveryEvents {
   PEER_DISCOVERY_BOOTSTRAP = "peer:discovery:bootstrap",
   PEER_DISCOVERY_PEER_EXCHANGE = "peer:discovery:peer-exchange",
   PEER_CONNECT_BOOTSTRAP = "peer:connected:bootstrap",
-  PEER_CONNECT_PEER_EXCHANGE = "peer:connected:peer-exchange"
+  PEER_CONNECT_PEER_EXCHANGE = "peer:connected:peer-exchange",
+  NODE_ONLINE = "waku:online",
+  NODE_OFFLINE = "waku:offline"
 }
 
 export interface IPeersByDiscoveryEvents {
   [EPeersByDiscoveryEvents.PEER_DISCOVERY_BOOTSTRAP]: CustomEvent<PeerId>;
   [EPeersByDiscoveryEvents.PEER_DISCOVERY_PEER_EXCHANGE]: CustomEvent<PeerId>;
   [EPeersByDiscoveryEvents.PEER_CONNECT_BOOTSTRAP]: CustomEvent<PeerId>;
   [EPeersByDiscoveryEvents.PEER_CONNECT_PEER_EXCHANGE]: CustomEvent<PeerId>;
+  [EPeersByDiscoveryEvents.NODE_ONLINE]: CustomEvent<void>;
+  [EPeersByDiscoveryEvents.NODE_OFFLINE]: CustomEvent<void>;
 }
 
 export interface PeersByDiscoveryResult {

packages/tests/tests/connection_manager.spec.ts

@@ -156,6 +156,97 @@ describe("ConnectionManager", function () {
         expect(await peerConnectedPeerExchange).to.eq(true);
       });
     });
+    it("should emit `waku:online` event only when first peer is connected", async function () {
+      const peerIdPx = await createSecp256k1PeerId();
+
+      await waku.libp2p.peerStore.save(peerIdPx, {
+        tags: {
+          [Tags.PEER_EXCHANGE]: {
+            value: 50,
+            ttl: 1200000
+          }
+        }
+      });
+
+      let eventCount = 0;
+      const wakuOnline = new Promise<boolean>((resolve) => {
+        waku.connectionManager.addEventListener(
+          EPeersByDiscoveryEvents.NODE_ONLINE,
+          () => {
+            eventCount++;
+            resolve(true);
+          }
+        );
+      });
+
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:connect", { detail: peerIdPx })
+      );
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:connect", {
+          detail: await createSecp256k1PeerId()
+        })
+      );
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:connect", {
+          detail: await createSecp256k1PeerId()
+        })
+      );
+
+      expect(await wakuOnline).to.eq(true);
+      expect(eventCount).to.be.eq(1);
+    });
+    // This test currently fails
+    it("should emit `waku:offline` event when all peers disconnect", async function () {
+      const peerIdPx = await createSecp256k1PeerId();
+      const peerIdPx2 = await createSecp256k1PeerId();
+
+      await waku.libp2p.peerStore.save(peerIdPx, {
+        tags: {
+          [Tags.PEER_EXCHANGE]: {
+            value: 50,
+            ttl: 1200000
+          }
+        }
+      });
+
+      await waku.libp2p.peerStore.save(peerIdPx2, {
+        tags: {
+          [Tags.PEER_EXCHANGE]: {
+            value: 50,
+            ttl: 1200000
+          }
+        }
+      });
+
+      let eventCount = 0;
+      const wakuOffline = new Promise<boolean>((resolve) => {
+        waku.connectionManager.addEventListener(
+          EPeersByDiscoveryEvents.NODE_OFFLINE,
+          () => {
+            eventCount++;
+            resolve(true);
+          }
+        );
+      });
+
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:connect", { detail: peerIdPx })
+      );
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:connect", { detail: peerIdPx2 })
+      );
+
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:disconnect", { detail: peerIdPx })
+      );
+      waku.libp2p.dispatchEvent(
+        new CustomEvent<PeerId>("peer:disconnect", { detail: peerIdPx2 })
+      );
+
+      expect(await wakuOffline).to.eq(true);
+      expect(eventCount).to.be.eq(1);
+    });
   });
 
   describe("Dials", () => {