feat(filter): support handler cleanup

Author: weboko

packages/core/src/lib/filter/filter.spec.ts

@@ -0,0 +1,34 @@
+import type { Libp2p } from "@waku/interfaces";
+import { expect } from "chai";
+import sinon from "sinon";
+
+import { FilterCodecs, FilterCore } from "./filter.js";
+
+function mockLibp2p(unhandle: sinon.SinonStub): Libp2p {
+  return {
+    handle: sinon.stub().resolves(unhandle),
+    components: {
+      events: {
+        addEventListener: sinon.stub(),
+        removeEventListener: sinon.stub()
+      },
+      connectionManager: {
+        getConnections: sinon.stub().returns([])
+      }
+    }
+  } as unknown as Libp2p;
+}
+
+describe("FilterCore", () => {
+  it("registers handler and removes on stop", async () => {
+    const unhandle = sinon.stub().resolves();
+    const libp2p = mockLibp2p(unhandle);
+
+    const core = new FilterCore(async () => Promise.resolve(), libp2p);
+
+    await core.stop();
+
+    expect((libp2p.handle as any).calledOnceWith(FilterCodecs.PUSH)).to.be.true;
+    expect(unhandle.calledOnce).to.be.true;
+  });
+});

packages/core/src/lib/filter/filter.ts

@@ -37,6 +37,8 @@ type IncomingMessageHandler = (
 
 export class FilterCore {
   private streamManager: StreamManager;
+  private unregister?: () => void | Promise<void>;
+  private handlePromise: Promise<void | (() => void | Promise<void>)>;
 
   public readonly multicodec = FilterCodecs.SUBSCRIBE;
 
@@ -49,15 +51,30 @@ export class FilterCore {
       libp2p.components
     );
 
-    libp2p
-      .handle(FilterCodecs.PUSH, this.onRequest.bind(this), {
+    this.handlePromise = (
+      libp2p.handle(FilterCodecs.PUSH, this.onRequest.bind(this), {
         maxInboundStreams: 100
+      }) as unknown as Promise<() => void | Promise<void>>
+    )
+      .then((unhandle) => {
+        this.unregister = unhandle;
+        return unhandle;
       })
       .catch((e) => {
         log.error("Failed to register ", FilterCodecs.PUSH, e);
       });
   }
 
+  public async stop(): Promise<void> {
+    await this.handlePromise;
+
+    try {
+      await this.unregister?.();
+    } catch (e) {
+      log.error("Failed to unregister ", FilterCodecs.PUSH, e);
+    }
+  }
+
   public async subscribe(
     pubsubTopic: PubsubTopic,
     peerId: PeerId,

packages/interfaces/src/filter.ts

@@ -73,6 +73,11 @@ export type IFilter = {
    * await filter.subscribe(newDecoder, newCallback);
    */
   unsubscribeAll(): void;
+
+  /**
+   * Stops the Filter protocol and removes any registered handlers.
+   */
+  stop(): Promise<void>;
 };
 
 export type FilterProtocolOptions = {

packages/sdk/src/filter/filter.spec.ts

@@ -98,26 +98,27 @@ describe("Filter SDK", () => {
     expect(subscriptionInvokeStub.firstCall.args[1]).to.equal(peerId);
   });
 
-  it("should successfully stop", async () => {
+  it("should stop and remove handler", async () => {
     const contentTopic2 = "/test/1/waku-filter-2/utf8";
     const decoder2 = createDecoder(
       contentTopic2,
       createRoutingInfo(testNetworkconfig, { contentTopic: contentTopic2 })
     );
-    const stopStub = sinon.stub(Subscription.prototype, "stop");
-
+    const subscriptionStopStub = sinon.stub(Subscription.prototype, "stop");
     sinon.stub(Subscription.prototype, "add").resolves(true);
     sinon.stub(Subscription.prototype, "start");
 
     await filter.subscribe(decoder, callback);
     await filter.subscribe(decoder2, callback);
 
-    filter.unsubscribeAll();
+    const protocolStopStub = sinon
+      .stub((filter as any)["protocol"], "stop")
+      .resolves();
 
-    expect(stopStub.calledOnce).to.be.true;
+    await filter.stop();
 
-    const result = await filter.unsubscribe(decoder);
-    expect(result).to.be.false;
+    expect(subscriptionStopStub.calledOnce).to.be.true;
+    expect(protocolStopStub.calledOnce).to.be.true;
   });
 });
 

packages/sdk/src/filter/filter.ts

@@ -45,6 +45,11 @@ export class Filter implements IFilter {
     return this.protocol.multicodec;
   }
 
+  public async stop(): Promise<void> {
+    this.unsubscribeAll();
+    await this.protocol.stop();
+  }
+
   public unsubscribeAll(): void {
     for (const subscription of this.subscriptions.values()) {
       subscription.stop();

packages/sdk/src/waku/waku.ts

@@ -231,6 +231,7 @@ export class WakuNode implements IWaku {
     this._nodeStateLock = true;
 
     this.lightPush?.stop();
+    await this.filter?.stop();
     this.healthIndicator.stop();
     this.peerManager.stop();
     this.connectionManager.stop();