feat(events): add new events service

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

plugins/events-node/api-report-alpha.md

@@ -13,15 +13,15 @@ import { HttpPostIngressOptions } from '@backstage/plugin-events-node';
 export interface EventsExtensionPoint {
   // (undocumented)
   addHttpPostIngress(options: HttpPostIngressOptions): void;
-  // (undocumented)
+  // @deprecated (undocumented)
   addPublishers(
     ...publishers: Array<EventPublisher | Array<EventPublisher>>
   ): void;
-  // (undocumented)
+  // @deprecated (undocumented)
   addSubscribers(
     ...subscribers: Array<EventSubscriber | Array<EventSubscriber>>
   ): void;
-  // (undocumented)
+  // @deprecated (undocumented)
   setEventBroker(eventBroker: EventBroker): void;
 }
 

plugins/events-node/api-report.md

@@ -3,7 +3,21 @@
 > Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
 
 ```ts
+import { LoggerService } from '@backstage/backend-plugin-api';
+import { ServiceRef } from '@backstage/backend-plugin-api';
+
 // @public
+export class DefaultEventsService implements EventsService {
+  // (undocumented)
+  static create(options: { logger: LoggerService }): DefaultEventsService;
+  forPlugin(pluginId: string): EventsService;
+  // (undocumented)
+  publish(params: EventParams): Promise<void>;
+  // (undocumented)
+  subscribe(options: EventsServiceSubscribeOptions): Promise<void>;
+}
+
+// @public @deprecated
 export interface EventBroker {
   publish(params: EventParams): Promise<void>;
   subscribe(
@@ -18,9 +32,9 @@ export interface EventParams<TPayload = unknown> {
   topic: string;
 }
 
-// @public
+// @public @deprecated
 export interface EventPublisher {
-  // (undocumented)
+  // @deprecated (undocumented)
   setEventBroker(eventBroker: EventBroker): Promise<void>;
 }
 
@@ -39,8 +53,29 @@ export abstract class EventRouter implements EventPublisher, EventSubscriber {
 }
 
 // @public
+export interface EventsService {
+  publish(params: EventParams): Promise<void>;
+  subscribe(options: EventsServiceSubscribeOptions): Promise<void>;
+}
+
+// @public (undocumented)
+export type EventsServiceEventHandler = (params: EventParams) => Promise<void>;
+
+// @public
+export const eventsServiceRef: ServiceRef<EventsService, 'plugin'>;
+
+// @public (undocumented)
+export type EventsServiceSubscribeOptions = {
+  id: string;
+  topics: string[];
+  onEvent: EventsServiceEventHandler;
+};
+
+// @public @deprecated
 export interface EventSubscriber {
+  // @deprecated
   onEvent(params: EventParams): Promise<void>;
+  // @deprecated
   supportsEventTopics(): string[];
 }
 

plugins/events-node/package.json

@@ -45,6 +45,7 @@
     "@backstage/backend-plugin-api": "workspace:^"
   },
   "devDependencies": {
+    "@backstage/backend-common": "workspace:^",
     "@backstage/cli": "workspace:^"
   },
   "files": [

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

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2022 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { getVoidLogger } from '@backstage/backend-common';
+import { DefaultEventsService } from './DefaultEventsService';
+import { EventParams } from './EventParams';
+
+const logger = getVoidLogger();
+
+describe('DefaultEventsService', () => {
+  it('passes events to interested subscribers', async () => {
+    const events = DefaultEventsService.create({ logger });
+    const eventsSubscriber1: EventParams[] = [];
+    const eventsSubscriber2: EventParams[] = [];
+
+    await events.subscribe({
+      id: 'subscriber1',
+      topics: ['topicA', 'topicB'],
+      onEvent: async event => {
+        eventsSubscriber1.push(event);
+      },
+    });
+    await events.subscribe({
+      id: 'subscriber2',
+      topics: ['topicB', 'topicC'],
+      onEvent: async event => {
+        eventsSubscriber2.push(event);
+      },
+    });
+    await events.publish({
+      topic: 'topicA',
+      eventPayload: { test: 'topicA' },
+    });
+    await events.publish({
+      topic: 'topicB',
+      eventPayload: { test: 'topicB' },
+    });
+    await events.publish({
+      topic: 'topicC',
+      eventPayload: { test: 'topicC' },
+    });
+    await events.publish({
+      topic: 'topicD',
+      eventPayload: { test: 'topicD' },
+    });
+
+    expect(eventsSubscriber1).toEqual([
+      { topic: 'topicA', eventPayload: { test: 'topicA' } },
+      { topic: 'topicB', eventPayload: { test: 'topicB' } },
+    ]);
+    expect(eventsSubscriber2).toEqual([
+      { topic: 'topicB', eventPayload: { test: 'topicB' } },
+      { topic: 'topicC', eventPayload: { test: 'topicC' } },
+    ]);
+  });
+
+  it('logs errors from subscribers', async () => {
+    const topic = 'testTopic';
+
+    const warnSpy = jest.spyOn(logger, 'warn');
+    const events = DefaultEventsService.create({ logger });
+
+    await events.subscribe({
+      id: 'subscriber1',
+      topics: [topic],
+      onEvent: event => {
+        throw new Error(`NOPE ${event.eventPayload}`);
+      },
+    });
+    await events.publish({ topic, eventPayload: '1' });
+
+    expect(warnSpy).toHaveBeenCalledTimes(1);
+    expect(warnSpy).toHaveBeenCalledWith(
+      'Subscriber "subscriber1" failed to process event for topic "testTopic"',
+      new Error('NOPE 1'),
+    );
+
+    await events.subscribe({
+      id: 'subscriber2',
+      topics: [topic],
+      onEvent: event => {
+        throw new Error(`NOPE ${event.eventPayload}`);
+      },
+    });
+    await events.publish({ topic, eventPayload: '2' });
+
+    // With two subscribers we should not halt on the first error but call all subscribers
+    expect(warnSpy).toHaveBeenCalledTimes(3);
+    expect(warnSpy).toHaveBeenCalledWith(
+      'Subscriber "subscriber1" failed to process event for topic "testTopic"',
+      new Error('NOPE 2'),
+    );
+    expect(warnSpy).toHaveBeenCalledWith(
+      'Subscriber "subscriber2" failed to process event for topic "testTopic"',
+      new Error('NOPE 2'),
+    );
+  });
+});

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

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { LoggerService } from '@backstage/backend-plugin-api';
+import { EventParams } from './EventParams';
+import { EventsService, EventsServiceSubscribeOptions } from './EventsService';
+
+/**
+ * In-process event broker which will pass the event to all registered subscribers
+ * interested in it.
+ * Events will not be persisted in any form.
+ * Events will not be passed to subscribers at other instances of the same cluster.
+ *
+ * @public
+ */
+// TODO(pjungermann): add opentelemetry? (see plugins/catalog-backend/src/util/opentelemetry.ts, etc.)
+export class DefaultEventsService implements EventsService {
+  private readonly subscribers = new Map<
+    string,
+    Omit<EventsServiceSubscribeOptions, 'topics'>[]
+  >();
+
+  private constructor(private readonly logger: LoggerService) {}
+
+  static create(options: { logger: LoggerService }): DefaultEventsService {
+    return new DefaultEventsService(options.logger);
+  }
+
+  /**
+   * Returns a plugin-scoped context of the `EventService`
+   * that ensures to prefix subscriber IDs with the plugin ID.
+   *
+   * @param pluginId - The plugin that the `EventService` should be created for.
+   */
+  forPlugin(pluginId: string): EventsService {
+    return {
+      publish: (params: EventParams): Promise<void> => {
+        return this.publish(params);
+      },
+      subscribe: (options: EventsServiceSubscribeOptions): Promise<void> => {
+        return this.subscribe({
+          ...options,
+          id: `${pluginId}.${options.id}`,
+        });
+      },
+    };
+  }
+
+  async publish(params: EventParams): Promise<void> {
+    this.logger.debug(
+      `Event received: topic=${params.topic}, metadata=${JSON.stringify(
+        params.metadata,
+      )}, payload=${JSON.stringify(params.eventPayload)}`,
+    );
+
+    if (!this.subscribers.has(params.topic)) {
+      return;
+    }
+
+    const onEventPromises: Promise<void>[] = [];
+    this.subscribers.get(params.topic)?.forEach(subscription => {
+      onEventPromises.push(
+        (async () => {
+          try {
+            await subscription.onEvent(params);
+          } catch (error) {
+            this.logger.warn(
+              `Subscriber "${subscription.id}" failed to process event for topic "${params.topic}"`,
+              error,
+            );
+          }
+        })(),
+      );
+    });
+
+    await Promise.all(onEventPromises);
+  }
+
+  async subscribe(options: EventsServiceSubscribeOptions): Promise<void> {
+    options.topics.forEach(topic => {
+      if (!this.subscribers.has(topic)) {
+        this.subscribers.set(topic, []);
+      }
+
+      this.subscribers.get(topic)!.push({
+        id: options.id,
+        onEvent: options.onEvent,
+      });
+    });
+  }
+}

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

@@ -23,6 +23,7 @@ import { EventSubscriber } from './EventSubscriber';
  * others can subscribe for future events for topics they are interested in.
  *
  * @public
+ * @deprecated use `EventsService` instead
  */
 export interface EventBroker {
   /**

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

@@ -23,7 +23,11 @@ import { EventBroker } from './EventBroker';
  * or from event brokers, queues, etc.
  *
  * @public
+ * @deprecated use the `EventsService` via the constructor, setter, or other means instead
  */
 export interface EventPublisher {
+  /**
+   * @deprecated use the `EventsService` via the constructor, setter, or other means instead
+   */
   setEventBroker(eventBroker: EventBroker): Promise<void>;
 }

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

@@ -22,10 +22,13 @@ import { EventParams } from './EventParams';
  * or other actions to react on events.
  *
  * @public
+ * @deprecated use the `EventsService` via the constructor, setter, or other means instead
  */
 export interface EventSubscriber {
   /**
    * Supported event topics like "github", "bitbucketCloud", etc.
+   *
+   * @deprecated use the `EventsService` via the constructor, setter, or other means instead
    */
   supportsEventTopics(): string[];
 
@@ -33,6 +36,7 @@ export interface EventSubscriber {
    * React on a received event.
    *
    * @param params - parameters for the to be received event.
+   * @deprecated you are not required to expose this anymore when using `EventsService`
    */
   onEvent(params: EventParams): Promise<void>;
 }

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { EventParams } from './EventParams';
+
+/**
+ * Allows a decoupled and asynchronous communication between components.
+ * Components can publish events for a given topic and
+ * others can subscribe for future events for topics they are interested in.
+ *
+ * @public
+ */
+export interface EventsService {
+  /**
+   * Publishes an event for the topic.
+   *
+   * @param params - parameters for the to be published event.
+   */
+  publish(params: EventParams): Promise<void>;
+
+  /**
+   * Subscribes to one or more topics, registering an event handler for them.
+   *
+   * @param options - event subscription options.
+   */
+  subscribe(options: EventsServiceSubscribeOptions): Promise<void>;
+}
+
+/**
+ * @public
+ */
+export type EventsServiceSubscribeOptions = {
+  /**
+   * Identifier for the subscription. E.g., used as part of log messages.
+   */
+  id: string;
+  topics: string[];
+  onEvent: EventsServiceEventHandler;
+};
+
+/**
+ * @public
+ */
+export type EventsServiceEventHandler = (params: EventParams) => Promise<void>;

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

@@ -14,10 +14,13 @@
  * limitations under the License.
  */
 
-export type { EventBroker } from './EventBroker';
 export type { EventParams } from './EventParams';
-export type { EventPublisher } from './EventPublisher';
 export { EventRouter } from './EventRouter';
-export type { EventSubscriber } from './EventSubscriber';
+export type {
+  EventsService,
+  EventsServiceSubscribeOptions,
+  EventsServiceEventHandler,
+} from './EventsService';
+export { DefaultEventsService } from './DefaultEventsService';
 export * from './http';
 export { SubTopicEventRouter } from './SubTopicEventRouter';

plugins/events-node/src/deprecated.ts

@@ -0,0 +1,19 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export type { EventBroker } from './api/EventBroker';
+export type { EventPublisher } from './api/EventPublisher';
+export type { EventSubscriber } from './api/EventSubscriber';

plugins/events-node/src/extensions.ts

@@ -26,12 +26,21 @@ import {
  * @alpha
  */
 export interface EventsExtensionPoint {
+  /**
+   * @deprecated use `eventsServiceRef` and `eventsServiceFactory` instead
+   */
   setEventBroker(eventBroker: EventBroker): void;
 
+  /**
+   * @deprecated use `EventsService.publish` instead
+   */
   addPublishers(
     ...publishers: Array<EventPublisher | Array<EventPublisher>>
   ): void;
 
+  /**
+   * @deprecated use `EventsService.subscribe` instead
+   */
   addSubscribers(
     ...subscribers: Array<EventSubscriber | Array<EventSubscriber>>
   ): void;

plugins/events-node/src/index.ts

@@ -21,3 +21,5 @@
  */
 
 export * from './api';
+export * from './deprecated';
+export { eventsServiceRef } from './service';

plugins/events-node/src/service.ts

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {
+  coreServices,
+  createServiceFactory,
+  createServiceRef,
+} from '@backstage/backend-plugin-api';
+import { EventsService, DefaultEventsService } from './api';
+
+/**
+ * The {@link EventsService} that allows to publish events, and subscribe to topics.
+ * Uses the `root` scope so that events can be shared across all plugins, modules, and more.
+ *
+ * @public
+ */
+export const eventsServiceRef = createServiceRef<EventsService>({
+  id: 'events.service',
+  scope: 'plugin',
+  defaultFactory: async service =>
+    createServiceFactory({
+      service,
+      deps: {
+        pluginMetadata: coreServices.pluginMetadata,
+        rootLogger: coreServices.rootLogger,
+      },
+      async createRootContext({ rootLogger }) {
+        return DefaultEventsService.create({ logger: rootLogger });
+      },
+      async factory({ pluginMetadata }, eventsService) {
+        return eventsService.forPlugin(pluginMetadata.getId());
+      },
+    }),
+});