feat: support publish async from worker

Author: czy88840616

src/base.ts

@@ -391,9 +391,19 @@ export abstract class AbstractEventBus<T> implements IEventBus<T> {
           const isChunk = originMessage.messageOptions?.['isChunk'] === true;
           const responder = new AckResponder();
           responder.onData(data => {
-            this.publish(data, {
-              relatedMessageId: originMessage.messageId,
-              isChunk,
+            this.transit({
+              messageCategory: MessageCategory.OUT,
+              message: {
+                messageId: this.generateMessageId(),
+                workerId: this.getWorkerId(),
+                type: MessageType.Response,
+                body: data,
+                messageOptions: {
+                  relatedMessageId: originMessage.messageId,
+                  isChunk,
+                  topic: originMessage.messageOptions?.topic,
+                },
+              },
             });
 
             if (!isChunk) {
@@ -586,7 +596,7 @@ export abstract class AbstractEventBus<T> implements IEventBus<T> {
         message: {
           messageId,
           workerId: this.getWorkerId(),
-          type: this.isMain() ? MessageType.Invoke : MessageType.Response,
+          type: MessageType.Invoke,
           body: data,
           messageOptions: {
             topic: publishOptions.topic,

test/cp.test.ts

@@ -545,4 +545,27 @@ describe('/test/cp.test.ts', function () {
     worker.kill();
     await bus.stop();
   });
+
+  it('test publishAsync from child to main', async () => {
+    const bus = new ChildProcessEventBus();
+    const worker = createChildProcessWorker(join(__dirname, 'cp/publish_async_from_worker.ts'));
+    bus.addWorker(worker);
+    await bus.start();
+
+    await new Promise<void>(resolve => {
+      // 订阅来自子进程的消息，并返回响应
+      bus.subscribe((message, responder?) => {
+        if (message.body.data === 'request from child') {
+          responder?.send({
+            data: 'response from main',
+          });
+        } else if (message.body.data === 'ok') {
+          resolve();
+        }
+      });
+    });
+
+    await worker.kill();
+    await bus.stop();
+  });
 });

test/cp/publish_async_from_worker.ts

@@ -0,0 +1,24 @@
+import { ChildProcessEventBus } from '../../src';
+
+async function createWorker() {
+  const bus = new ChildProcessEventBus({
+    isWorker: true,
+  });
+
+  await bus.start();
+
+  // 在子进程中发送异步消息到主进程
+  const result = await bus.publishAsync<{data: string}>({
+    data: 'request from child',
+  });
+
+  if (result.data === 'response from main') {
+    bus.publish({
+      data: 'ok',
+    });
+  }
+}
+
+createWorker().then(() => {
+  console.log('Child process is ready');
+});

test/local.test.ts

@@ -22,7 +22,7 @@ describe('/test/local.test.ts', function () {
     await bus.stop();
   });
 
-  it.only('test base publish and subscribe', async () => {
+  it('test base publish and subscribe', async () => {
     const bus = new LocalEventBus({
       isWorker: false,
     });
@@ -49,7 +49,7 @@ describe('/test/local.test.ts', function () {
     await bus.stop();
   });
 
-  it.only('test base subscribe and abort', async () => {
+  it('test base subscribe and abort', async () => {
     const bus = new LocalEventBus({
       isWorker: false,
     });

test/local/publish_async_from_worker.ts

@@ -0,0 +1,24 @@
+import { LocalEventBus } from '../../src';
+
+async function createWorker() {
+  const bus = new LocalEventBus({
+    isWorker: true,
+  });
+
+  await bus.start();
+
+  // 在子进程中发送异步消息到主进程
+  const result = await bus.publishAsync<{data: string}>({
+    data: 'request from child',
+  });
+
+  if (result.data === 'response from main') {
+    bus.publish({
+      data: 'ok',
+    });
+  }
+}
+
+createWorker().then(() => {
+  console.log('Local worker is ready');
+}); 

test/worker/publish_async_from_worker.ts

@@ -0,0 +1,24 @@
+import { ThreadEventBus } from '../../src';
+
+async function createWorker() {
+  const bus = new ThreadEventBus({
+    isWorker: true,
+  });
+
+  await bus.start();
+
+  // 在子进程中发送异步消息到主进程
+  const result = await bus.publishAsync<{data: string}>({
+    data: 'request from child',
+  });
+
+  if (result.data === 'response from main') {
+    bus.publish({
+      data: 'ok',
+    });
+  }
+}
+
+createWorker().then(() => {
+  console.log('Thread worker is ready');
+});