feat: add abort controller for subscribe

src/base.ts

@@ -10,6 +10,7 @@
   MessageCategory,
   MessageType,
   PublishOptions,
+  SubscribeAbortController,
   SubscribeOptions,
   SubscribeTopicListener,
   WaitCheckOptions,
@@ -320,7 +321,7 @@
     } else {
       this.debugLogger(`Skip init worker(${this.getWorkerId(worker)}) status`);
     }
     worker?.['on']('exit', async (exitCode: number) => {
       if (!this.stopping) {
         // remove ready status
         this.workerReady.delete(this.getWorkerId(worker));
@@ -493,7 +494,7 @@
   public subscribe(
     listener: SubscribeTopicListener,
     options: SubscribeOptions = {}
-  ) {
+  ): SubscribeAbortController {
     const topic = options.topic || DEFAULT_LISTENER_KEY;
     if (!this.topicListener.has(topic)) {
       this.topicListener.set(topic, new Set());
@@ -502,6 +503,11 @@
       listener['_subscribeOnce'] = true;
     }
     this.topicListener.get(topic).add(listener);
+    listener['_abortController'] = {
+      abort: () => {
+        this.topicListener.get(topic).delete(listener);
+      },
+    };
 
     // if topic has cache
     if (
@@ -517,14 +523,16 @@
       }
       this.topicMessageCache.get(topic).clear();
     }
+
+    return listener['_abortController'];
   }
 
   public subscribeOnce(
     listener: SubscribeTopicListener,
     options: SubscribeOptions = {}
-  ) {
+  ): SubscribeAbortController {
     options.subscribeOnce = true;
-    this.subscribe(listener, options);
+    return this.subscribe(listener, options);
   }
 
   public publish(

src/interface.ts

@@ -122,14 +122,21 @@ export type SubscribeTopicListener = (
   responder?: IResponder
 ) => void | Promise<void>;
 
+export type SubscribeAbortController = {
+  abort: () => void;
+};
+
 export interface IEventBus<T> {
   addWorker(worker: T);
   start(err?: Error): Promise<void>;
-  subscribe(callback: SubscribeTopicListener, options?: SubscribeOptions): void;
+  subscribe(
+    callback: SubscribeTopicListener,
+    options?: SubscribeOptions
+  ): SubscribeAbortController;
   subscribeOnce(
     callback: SubscribeTopicListener,
     options?: SubscribeOptions
-  ): void;
+  ): SubscribeAbortController;
   publishAsync<ResData>(
     data: unknown,
     publishOptions?: PublishOptions

test/cp.test.ts

@@ -51,6 +51,38 @@ describe('/test/cp.test.ts', function () {
     await bus.stop();
   });
 
+  it('test base subscribe with abort', async () => {
+    const bus = new ChildProcessEventBus();
+    const worker = createChildProcessWorker(join(__dirname, 'cp/base.ts'));
+    bus.addWorker(worker);
+    await bus.start();
+
+    const result = await new Promise(resolve => {
+      const abortController = bus.subscribe(message => {
+        resolve(message.body);
+      });
+
+      abortController.abort();
+
+      bus.publish({
+        data: {
+          name: 'test',
+        }
+      }, {
+        topic: 'target'
+      });
+
+      setTimeout(() => {
+        resolve(null);
+      }, 1000);
+    });
+
+    expect(result).toEqual(null);
+
+    worker.kill();
+    await bus.stop();
+  });
+
   it('test publish with async', async () => {
     const bus = new ChildProcessEventBus();
     const worker = createChildProcessWorker(join(__dirname, 'cp/publish_async.ts'));

test/local.test.ts

@@ -22,7 +22,7 @@ describe('/test/local.test.ts', function () {
     await bus.stop();
   });
 
-  it('test base publish and subscribe', async () => {
+  it.only('test base publish and subscribe', async () => {
     const bus = new LocalEventBus({
       isWorker: false,
     });
@@ -49,6 +49,39 @@ describe('/test/local.test.ts', function () {
     await bus.stop();
   });
 
+  it.only('test base subscribe and abort', async () => {
+    const bus = new LocalEventBus({
+      isWorker: false,
+    });
+    createLocalWorker(join(__dirname, 'local/base_abort.ts'));
+    await bus.start();
+
+    const result = await new Promise(resolve => {
+      const abortController = bus.subscribe(message => {
+        resolve(message.body);
+      });
+
+      abortController.abort();
+
+      bus.publish({
+          data: {
+            name: 'test',
+          }
+        },
+        {
+          topic: 'target',
+        });
+
+      setTimeout(() => {
+        resolve(null);
+      }, 1000);
+    });
+
+    expect(result).toEqual(null);
+
+    await bus.stop();
+  });
+
   it('test publish with async', async () => {
     const bus = new LocalEventBus({
       isWorker: false,

test/local/base_abort.ts

@@ -0,0 +1,24 @@
+import { LocalEventBus } from '../../src/index';
+
+export async function createWorker() {
+  const bus = new LocalEventBus({
+    isWorker: true,
+  });
+
+  bus.subscribe(message=>{
+    console.log(message);
+
+    bus.publish({
+      data: 'hello world'
+    });
+  },
+  {
+    topic: 'target',
+  });
+
+  await bus.start();
+}
+
+createWorker().then(() => {
+  console.log('ready');
+});

test/thread.test.ts

@@ -52,6 +52,40 @@ describe('/test/thread.test.ts', function () {
     await bus.stop();
   });
 
+  it('test base subscribe and abort', async () => {
+    const bus = new ThreadEventBus();
+    const worker = createThreadWorker(join(__dirname, 'worker/base.ts'));
+    bus.addWorker(worker);
+    await bus.start();
+
+    const result = await new Promise(resolve => {
+      const abortController = bus.subscribe(message => {
+        resolve(message.body);
+      });
+
+      abortController.abort();
+
+      bus.publish({
+          data: {
+            name: 'test',
+          }
+        },
+        {
+          topic: 'target',
+        });
+
+      setTimeout(() => {
+        resolve(null);
+      }, 1000);
+
+    });
+
+    expect(result).toEqual(null);
+
+    await worker.terminate();
+    await bus.stop();
+  });
+
   it('test publish with async', async () => {
     const bus = new ThreadEventBus();
     const worker = createThreadWorker(join(__dirname, 'worker/publish_async.ts'));