merge MessageBus and PubSub interfaces

packages/cqrs/src/__tests__/iterator.spec.ts

@@ -1,6 +1,6 @@
 import { ErrorName, immediate } from '@mithic/commons';
-import { SimpleMessageBus } from '../bus/simple.js';
 import { AsyncSubscriber } from '../iterator.js';
+import { SimpleMessageBus } from '@mithic/messaging';
 
 describe(AsyncSubscriber.name, () => {
   let eventBus: SimpleMessageBus<number>;
@@ -17,7 +17,7 @@ describe(AsyncSubscriber.name, () => {
 
   it('should return values from the subscription', async () => {
     const events = [1, 2, 3];
-    events.forEach(eventBus.dispatch);
+    events.forEach((event) => eventBus.dispatch(event));
     const result = [];
 
     for await (const event of subscriber) {
@@ -36,7 +36,7 @@ describe(AsyncSubscriber.name, () => {
 
     const error = new Error('stop');
     const events = [1, 2, 3];
-    events.forEach(eventBus.dispatch);
+    events.forEach((event) => eventBus.dispatch(event));
     const result = [];
 
     try {
@@ -60,7 +60,7 @@ describe(AsyncSubscriber.name, () => {
     const controller = new AbortController();
     const subscriber = new AsyncSubscriber(eventBus, controller);
     const events = [1, 2, 3];
-    events.forEach(eventBus.dispatch);
+    events.forEach((event) => eventBus.dispatch(event));
     const result = [];
 
     try {
@@ -96,7 +96,7 @@ describe(AsyncSubscriber.name, () => {
 
   it('should resolve pending pulls on close', async () => {
     const events = [1, 2];
-    events.forEach(eventBus.dispatch);
+    events.forEach((event) => eventBus.dispatch(event));
     const result = [];
     for (let i = 0; i < 4; ++i) {
       result.push(subscriber.next());
@@ -113,7 +113,7 @@ describe(AsyncSubscriber.name, () => {
   it('should drop overflowing values', async () => {
     const subscriber = new AsyncSubscriber(eventBus, { bufferSize: 1 });
     const events = [1, 2, 3];
-    events.forEach(eventBus.dispatch);
+    events.forEach((event) => eventBus.dispatch(event));
 
     for await (const event of subscriber) {
       expect(event).toEqual(events[2]); // previous events dropped
@@ -124,7 +124,7 @@ describe(AsyncSubscriber.name, () => {
   it('should ignore new values on fcfs mode if buffer is full', async () => {
     const subscriber = new AsyncSubscriber(eventBus, { bufferSize: 1, fcfs: true });
     const events = [1, 2, 3];
-    events.forEach(eventBus.dispatch);
+    events.forEach((event) => eventBus.dispatch(event));
 
     for await (const event of subscriber) {
       expect(event).toEqual(events[0]); // later events dropped

packages/cqrs/src/__tests__/mocks.ts

@@ -1,22 +1,24 @@
-import { MessageHandler, PubSub, PubSubMessage, SubscribeOptions, MessageValidator } from '@mithic/messaging';
+import { MessageHandler, SubscribeOptions, MessageValidator, MessageBus, Unsubscribe } from '@mithic/messaging';
 
 export const LIBP2P_KEY_CODE = 0x72;
 
-export class MockPubSub implements PubSub<Uint8Array> {
-  topicHandlers = new Map<string, MessageHandler<PubSubMessage<Uint8Array>>>();
-  topicValidators = new Map<string, MessageValidator<PubSubMessage<Uint8Array>>>();
+export class MockMessageBus<Msg = Uint8Array> implements MessageBus<Msg> {
+  topicHandlers = new Map<string, MessageHandler<Msg>>();
+  topicValidators = new Map<string, MessageValidator<Msg>>();
 
-  subscribe(topic: string, handler: MessageHandler<PubSubMessage<Uint8Array>>, options?: SubscribeOptions<PubSubMessage<Uint8Array>>): void {
+  subscribe(handler: MessageHandler<Msg>, options?: SubscribeOptions<Msg>): Unsubscribe {
+    const topic = options?.topic ?? '';
     this.topicHandlers.set(topic, handler);
     options?.validator && this.topicValidators.set(topic, options.validator);
+    return () => this.unsubscribe(topic);
   }
 
   unsubscribe(topic: string): void {
     this.topicHandlers.delete(topic);
     this.topicValidators.delete(topic);
   }
 
-  publish(): void {
+  dispatch(): void {
     // NO-OP
   }
 

packages/cqrs/src/__tests__/processor.spec.ts

@@ -1,11 +1,10 @@
 import { jest } from '@jest/globals';
-import { MessageConsumer, MessageSubscription, Unsubscribe } from '../bus.js';
-import { SimpleMessageBus } from '../bus/index.js';
 import { MessageProcessor } from '../processor.js';
+import { MessageHandler, MessageSubscription, SimpleMessageBus, Unsubscribe } from '@mithic/messaging';
 
 describe(MessageProcessor.name, () => {
   let subscription: MessageSubscription<string>;
-  let mockConsumer: jest.MockedFunction<MessageConsumer<string>>;
+  let mockConsumer: jest.MockedFunction<MessageHandler<string>>;
   let processor: MessageProcessor<string>;
 
   beforeEach(() => {

packages/cqrs/src/index.ts

@@ -1,8 +1,6 @@
-export * from './bus.js';
 export * from './event.js';
 export * from './iterator.js';
 export * from './processor.js';
 
-export * from './bus/index.js';
 export * from './processor/index.js';
 export * from './preset/index.js';

packages/cqrs/src/iterator.ts

@@ -1,6 +1,6 @@
 import { ArrayDeque } from '@mithic/collections';
 import { AbortOptions, Closeable, MaybePromise } from '@mithic/commons';
-import { MessageSubscription, Unsubscribe } from './bus.js';
+import { MessageSubscription, Unsubscribe } from '@mithic/messaging';
 
 /**
  * Subscribe to an {@link MessageSubscription} as AsyncIterator.