feat(events)!: migrate EventRouter implementations from EventBroker to EventsService

Signed-off-by: Patrick Jungermann <Patrick.Jungermann@gmail.com>

plugins/events-node/src/api/EventRouter.test.ts

@@ -14,11 +14,19 @@
  * limitations under the License.
  */
 
-import { EventBroker } from './EventBroker';
 import { EventParams } from './EventParams';
 import { EventRouter } from './EventRouter';
+import { EventsService } from './EventsService';
 
 class TestEventRouter extends EventRouter {
+  constructor(events: EventsService) {
+    super({ events, topics: ['my-topic'] });
+  }
+
+  protected getSubscriberId(): string {
+    return 'TestEventRouter';
+  }
+
   protected determineDestinationTopic(params: EventParams): string | undefined {
     const payload = params.eventPayload as { value?: number };
     if (payload.value === undefined) {
@@ -27,26 +35,21 @@ class TestEventRouter extends EventRouter {
 
     return payload.value % 2 === 0 ? 'even' : 'odd';
   }
-
-  supportsEventTopics(): string[] {
-    return ['my-topic'];
-  }
 }
 
 describe('EventRouter', () => {
-  const eventRouter = new TestEventRouter();
+  const published: EventParams[] = [];
+  const events: EventsService = {
+    publish: async event => {
+      published.push(event);
+    },
+    subscribe: async _subscription => {},
+  };
+  const eventRouter = new TestEventRouter(events);
   const topic = 'my-topic';
   const metadata = { random: 'metadata' };
 
   it('no destination topic', async () => {
-    const published: EventParams[] = [];
-    const eventBroker = {
-      publish: (params: EventParams) => {
-        published.push(params);
-      },
-    } as EventBroker;
-    await eventRouter.setEventBroker(eventBroker);
-
     await eventRouter.onEvent({
       topic,
       eventPayload: { discarded: 'event' },
@@ -57,14 +60,6 @@ describe('EventRouter', () => {
   });
 
   it('with destination topic', async () => {
-    const published: EventParams[] = [];
-    const eventBroker = {
-      publish: (params: EventParams) => {
-        published.push(params);
-      },
-    } as EventBroker;
-    await eventRouter.setEventBroker(eventBroker);
-
     const payloadEven = { value: 2 };
     const payloadOdd = { value: 3 };
     await eventRouter.onEvent({ topic, eventPayload: payloadEven, metadata });

plugins/events-node/src/api/EventRouter.ts

@@ -14,10 +14,8 @@
  * limitations under the License.
  */
 
-import { EventBroker } from './EventBroker';
 import { EventParams } from './EventParams';
-import { EventPublisher } from './EventPublisher';
-import { EventSubscriber } from './EventSubscriber';
+import { EventsService } from './EventsService';
 
 /**
  * Subscribes to a topic and - depending on a set of conditions -
@@ -26,13 +24,41 @@ import { EventSubscriber } from './EventSubscriber';
  * @see {@link https://www.enterpriseintegrationpatterns.com/MessageRouter.html | Message Router pattern}.
  * @public
  */
-export abstract class EventRouter implements EventPublisher, EventSubscriber {
-  private eventBroker?: EventBroker;
+export abstract class EventRouter {
+  private readonly events: EventsService;
+  private readonly topics: string[];
+  private subscribed: boolean = false;
+
+  protected constructor(options: { events: EventsService; topics: string[] }) {
+    this.events = options.events;
+    this.topics = options.topics;
+  }
+
+  protected abstract getSubscriberId(): string;
 
   protected abstract determineDestinationTopic(
     params: EventParams,
   ): string | undefined;
 
+  /**
+   * Subscribes itself to the topic(s),
+   * after which events potentially can be received
+   * and processed by {@link EventRouter.onEvent}.
+   */
+  async subscribe(): Promise<void> {
+    if (this.subscribed) {
+      return;
+    }
+
+    this.subscribed = true;
+
+    await this.events.subscribe({
+      id: this.getSubscriberId(),
+      topics: this.topics,
+      onEvent: this.onEvent.bind(this),
+    });
+  }
+
   async onEvent(params: EventParams): Promise<void> {
     const topic = this.determineDestinationTopic(params);
 
@@ -41,15 +67,9 @@ export abstract class EventRouter implements EventPublisher, EventSubscriber {
     }
 
     // republish to different topic
-    this.eventBroker?.publish({
+    await this.events.publish({
       ...params,
       topic,
     });
   }
-
-  async setEventBroker(eventBroker: EventBroker): Promise<void> {
-    this.eventBroker = eventBroker;
-  }
-
-  abstract supportsEventTopics(): string[];
 }

plugins/events-node/src/api/SubTopicEventRouter.test.ts

@@ -14,13 +14,17 @@
  * limitations under the License.
  */
 
-import { EventBroker } from './EventBroker';
 import { EventParams } from './EventParams';
+import { EventsService } from './EventsService';
 import { SubTopicEventRouter } from './SubTopicEventRouter';
 
 class TestSubTopicEventRouter extends SubTopicEventRouter {
-  constructor() {
-    super('my-topic');
+  constructor(events: EventsService) {
+    super({ events, topic: 'my-topic' });
+  }
+
+  protected getSubscriberId(): string {
+    return 'TestSubTopicEventRouter';
   }
 
   protected determineSubTopic(params: EventParams): string | undefined {
@@ -29,34 +33,25 @@ class TestSubTopicEventRouter extends SubTopicEventRouter {
 }
 
 describe('SubTopicEventRouter', () => {
-  const eventRouter = new TestSubTopicEventRouter();
+  const published: EventParams[] = [];
+  const events: EventsService = {
+    publish: async event => {
+      published.push(event);
+    },
+    subscribe: async _subscription => {},
+  };
+  const eventRouter = new TestSubTopicEventRouter(events);
   const topic = 'my-topic';
   const eventPayload = { test: 'payload' };
   const metadata = { 'x-my-event': 'test.type' };
 
   it('no x-my-event', async () => {
-    const published: EventParams[] = [];
-    const eventBroker = {
-      publish: (params: EventParams) => {
-        published.push(params);
-      },
-    } as EventBroker;
-    await eventRouter.setEventBroker(eventBroker);
-
     await eventRouter.onEvent({ topic, eventPayload });
 
     expect(published).toEqual([]);
   });
 
   it('with x-my-event', async () => {
-    const published: EventParams[] = [];
-    const eventBroker = {
-      publish: (params: EventParams) => {
-        published.push(params);
-      },
-    } as EventBroker;
-    await eventRouter.setEventBroker(eventBroker);
-
     await eventRouter.onEvent({ topic, eventPayload, metadata });
 
     expect(published.length).toBe(1);

plugins/events-node/src/api/SubTopicEventRouter.ts

@@ -16,6 +16,7 @@
 
 import { EventParams } from './EventParams';
 import { EventRouter } from './EventRouter';
+import { EventsService } from './EventsService';
 
 /**
  * Subscribes to the provided (generic) topic
@@ -27,8 +28,11 @@ import { EventRouter } from './EventRouter';
  * @public
  */
 export abstract class SubTopicEventRouter extends EventRouter {
-  protected constructor(private readonly topic: string) {
-    super();
+  protected constructor(options: { events: EventsService; topic: string }) {
+    super({
+      events: options.events,
+      topics: [options.topic],
+    });
   }
 
   protected abstract determineSubTopic(params: EventParams): string | undefined;
@@ -37,8 +41,4 @@ export abstract class SubTopicEventRouter extends EventRouter {
     const subTopic = this.determineSubTopic(params);
     return subTopic ? `${params.topic}.${subTopic}` : undefined;
   }
-
-  supportsEventTopics(): string[] {
-    return [this.topic];
-  }
 }