From 56969b6e550ff48c5c9cefab40958b50f8dbb99e Mon Sep 17 00:00:00 2001 From: Patrick Jungermann Date: Tue, 23 Jan 2024 18:30:14 +0100 Subject: [PATCH] feat(events): add new events service Signed-off-by: Patrick Jungermann --- .changeset/breezy-cycles-count.md | 39 ++++++ .../events-backend-test-utils/api-report.md | 20 +++- .../src/deprecated.ts | 19 +++ .../events-backend-test-utils/src/index.ts | 1 + .../src/testUtils/TestEventBroker.ts | 5 +- .../src/testUtils/TestEventPublisher.ts | 5 +- .../src/testUtils/TestEventSubscriber.ts | 5 +- .../src/testUtils/TestEventsService.ts | 48 ++++++++ .../src/testUtils/index.ts | 4 +- plugins/events-backend/api-report.md | 9 +- plugins/events-backend/src/deprecated.ts | 18 +++ plugins/events-backend/src/index.ts | 3 +- .../src/service/DefaultEventBroker.test.ts | 18 +-- .../src/service/DefaultEventBroker.ts | 50 ++++---- .../src/service/EventsBackend.ts | 1 + plugins/events-node/api-report-alpha.md | 6 +- plugins/events-node/api-report.md | 39 +++++- plugins/events-node/package.json | 1 + .../src/api/DefaultEventsService.test.ts | 111 ++++++++++++++++++ .../src/api/DefaultEventsService.ts | 104 ++++++++++++++++ plugins/events-node/src/api/EventBroker.ts | 1 + plugins/events-node/src/api/EventPublisher.ts | 4 + .../events-node/src/api/EventSubscriber.ts | 4 + plugins/events-node/src/api/EventsService.ts | 57 +++++++++ plugins/events-node/src/api/index.ts | 9 +- plugins/events-node/src/deprecated.ts | 19 +++ plugins/events-node/src/extensions.ts | 9 ++ plugins/events-node/src/index.ts | 2 + plugins/events-node/src/service.ts | 47 ++++++++ yarn.lock | 1 + 30 files changed, 600 insertions(+), 59 deletions(-) create mode 100644 .changeset/breezy-cycles-count.md create mode 100644 plugins/events-backend-test-utils/src/deprecated.ts create mode 100644 plugins/events-backend-test-utils/src/testUtils/TestEventsService.ts create mode 100644 plugins/events-backend/src/deprecated.ts create mode 100644 plugins/events-node/src/api/DefaultEventsService.test.ts create mode 100644 plugins/events-node/src/api/DefaultEventsService.ts create mode 100644 plugins/events-node/src/api/EventsService.ts create mode 100644 plugins/events-node/src/deprecated.ts create mode 100644 plugins/events-node/src/service.ts diff --git a/.changeset/breezy-cycles-count.md b/.changeset/breezy-cycles-count.md new file mode 100644 index 0000000000..81998d9e8b --- /dev/null +++ b/.changeset/breezy-cycles-count.md @@ -0,0 +1,39 @@ +--- +'@backstage/plugin-events-backend-test-utils': patch +'@backstage/plugin-events-backend': patch +'@backstage/plugin-events-node': patch +--- + +Add new `EventsService` as well as `eventsServiceRef` for the new backend system. + +**Summary:** + +- new: + `EventsService`, `eventsServiceRef`, `TestEventsService` +- deprecated: + `EventBroker`, `EventPublisher`, `EventSubscriber`, `DefaultEventBroker`, `EventsBackend`, + most parts of `EventsExtensionPoint` (alpha), + `TestEventBroker`, `TestEventPublisher`, `TestEventSubscriber` + +Add the `eventsServiceRef` as dependency to your backend plugins +or backend plugin modules. + +**Details:** + +The previous implementation using the `EventsExtensionPoint` was added in the early stages +of the new backend system and does not respect the plugin isolation. +This made it not compatible anymore with the new backend system. + +Additionally, the previous interfaces had some room for simplification, +supporting less exposure of internal concerns as well. + +Hereby, this change adds a new `EventsService` interface as replacement for the now deprecated `EventBroker`. +The new interface does not require any `EventPublisher` or `EventSubscriber` interfaces anymore. +Instead, it is expected that the `EventsService` gets passed into publishers and subscribers, +and used internally. There is no need to expose anything of that at their own interfaces. + +Most parts of `EventsExtensionPoint` (alpha) are deprecated as well and were not usable +(by other plugins or their modules) anyway. + +The `DefaultEventBroker` implementation is deprecated and wraps the new `DefaultEventsService` implementation. +Optionally, an instance can be passed as argument to allow mixed setups to operate alongside. diff --git a/plugins/events-backend-test-utils/api-report.md b/plugins/events-backend-test-utils/api-report.md index 46131c4244..9630c3d4e2 100644 --- a/plugins/events-backend-test-utils/api-report.md +++ b/plugins/events-backend-test-utils/api-report.md @@ -6,9 +6,11 @@ import { EventBroker } from '@backstage/plugin-events-node'; import { EventParams } from '@backstage/plugin-events-node'; import { EventPublisher } from '@backstage/plugin-events-node'; +import { EventsService } from '@backstage/plugin-events-node'; +import { EventsServiceSubscribeOptions } from '@backstage/plugin-events-node'; import { EventSubscriber } from '@backstage/plugin-events-node'; -// @public (undocumented) +// @public @deprecated (undocumented) export class TestEventBroker implements EventBroker { // (undocumented) publish(params: EventParams): Promise; @@ -22,7 +24,7 @@ export class TestEventBroker implements EventBroker { readonly subscribed: EventSubscriber[]; } -// @public (undocumented) +// @public @deprecated (undocumented) export class TestEventPublisher implements EventPublisher { // (undocumented) get eventBroker(): EventBroker | undefined; @@ -31,6 +33,20 @@ export class TestEventPublisher implements EventPublisher { } // @public (undocumented) +export class TestEventsService implements EventsService { + // (undocumented) + publish(params: EventParams): Promise; + // (undocumented) + get published(): EventParams[]; + // (undocumented) + reset(): void; + // (undocumented) + subscribe(options: EventsServiceSubscribeOptions): Promise; + // (undocumented) + get subscribed(): EventsServiceSubscribeOptions[]; +} + +// @public @deprecated (undocumented) export class TestEventSubscriber implements EventSubscriber { constructor(name: string, topics: string[]); // (undocumented) diff --git a/plugins/events-backend-test-utils/src/deprecated.ts b/plugins/events-backend-test-utils/src/deprecated.ts new file mode 100644 index 0000000000..15072dcfb4 --- /dev/null +++ b/plugins/events-backend-test-utils/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 { TestEventBroker } from './testUtils/TestEventBroker'; +export { TestEventPublisher } from './testUtils/TestEventPublisher'; +export { TestEventSubscriber } from './testUtils/TestEventSubscriber'; diff --git a/plugins/events-backend-test-utils/src/index.ts b/plugins/events-backend-test-utils/src/index.ts index efba5be0d0..da090488fc 100644 --- a/plugins/events-backend-test-utils/src/index.ts +++ b/plugins/events-backend-test-utils/src/index.ts @@ -20,4 +20,5 @@ * @packageDocumentation */ +export * from './deprecated'; export * from './testUtils'; diff --git a/plugins/events-backend-test-utils/src/testUtils/TestEventBroker.ts b/plugins/events-backend-test-utils/src/testUtils/TestEventBroker.ts index c697a6506f..78556cfc63 100644 --- a/plugins/events-backend-test-utils/src/testUtils/TestEventBroker.ts +++ b/plugins/events-backend-test-utils/src/testUtils/TestEventBroker.ts @@ -20,7 +20,10 @@ import { EventSubscriber, } from '@backstage/plugin-events-node'; -/** @public */ +/** + * @public + * @deprecated use `TestEventsService` instead + */ export class TestEventBroker implements EventBroker { readonly published: EventParams[] = []; readonly subscribed: EventSubscriber[] = []; diff --git a/plugins/events-backend-test-utils/src/testUtils/TestEventPublisher.ts b/plugins/events-backend-test-utils/src/testUtils/TestEventPublisher.ts index c1b2038afb..51bad11278 100644 --- a/plugins/events-backend-test-utils/src/testUtils/TestEventPublisher.ts +++ b/plugins/events-backend-test-utils/src/testUtils/TestEventPublisher.ts @@ -16,7 +16,10 @@ import { EventBroker, EventPublisher } from '@backstage/plugin-events-node'; -/** @public */ +/** + * @public + * @deprecated `EventPublisher` was replaced by `EventsService.publish` + */ export class TestEventPublisher implements EventPublisher { #eventBroker?: EventBroker; diff --git a/plugins/events-backend-test-utils/src/testUtils/TestEventSubscriber.ts b/plugins/events-backend-test-utils/src/testUtils/TestEventSubscriber.ts index ef5758b804..3db9023a9b 100644 --- a/plugins/events-backend-test-utils/src/testUtils/TestEventSubscriber.ts +++ b/plugins/events-backend-test-utils/src/testUtils/TestEventSubscriber.ts @@ -16,7 +16,10 @@ import { EventParams, EventSubscriber } from '@backstage/plugin-events-node'; -/** @public */ +/** + * @public + * @deprecated `EventSubscriber` was replaced by `EventsService.subscribe`. + */ export class TestEventSubscriber implements EventSubscriber { readonly name: string; readonly topics: string[]; diff --git a/plugins/events-backend-test-utils/src/testUtils/TestEventsService.ts b/plugins/events-backend-test-utils/src/testUtils/TestEventsService.ts new file mode 100644 index 0000000000..c87072711c --- /dev/null +++ b/plugins/events-backend-test-utils/src/testUtils/TestEventsService.ts @@ -0,0 +1,48 @@ +/* + * 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, + EventsService, + EventsServiceSubscribeOptions, +} from '@backstage/plugin-events-node'; + +/** @public */ +export class TestEventsService implements EventsService { + #published: EventParams[] = []; + #subscribed: EventsServiceSubscribeOptions[] = []; + + async publish(params: EventParams): Promise { + this.#published.push(params); + } + + async subscribe(options: EventsServiceSubscribeOptions): Promise { + this.#subscribed.push(options); + } + + get published(): EventParams[] { + return this.#published; + } + + get subscribed(): EventsServiceSubscribeOptions[] { + return this.#subscribed; + } + + reset(): void { + this.#published = []; + this.#subscribed = []; + } +} diff --git a/plugins/events-backend-test-utils/src/testUtils/index.ts b/plugins/events-backend-test-utils/src/testUtils/index.ts index a571ba3075..d9eb544628 100644 --- a/plugins/events-backend-test-utils/src/testUtils/index.ts +++ b/plugins/events-backend-test-utils/src/testUtils/index.ts @@ -14,6 +14,4 @@ * limitations under the License. */ -export { TestEventBroker } from './TestEventBroker'; -export { TestEventPublisher } from './TestEventPublisher'; -export { TestEventSubscriber } from './TestEventSubscriber'; +export { TestEventsService } from './TestEventsService'; diff --git a/plugins/events-backend/api-report.md b/plugins/events-backend/api-report.md index aeb6f9363d..9fffe719ba 100644 --- a/plugins/events-backend/api-report.md +++ b/plugins/events-backend/api-report.md @@ -7,14 +7,17 @@ import { Config } from '@backstage/config'; import { EventBroker } from '@backstage/plugin-events-node'; import { EventParams } from '@backstage/plugin-events-node'; import { EventPublisher } from '@backstage/plugin-events-node'; +import { EventsService } from '@backstage/plugin-events-node'; import { EventSubscriber } from '@backstage/plugin-events-node'; import express from 'express'; import { HttpPostIngressOptions } from '@backstage/plugin-events-node'; import { Logger } from 'winston'; +import { LoggerService } from '@backstage/backend-plugin-api'; -// @public +// @public @deprecated export class DefaultEventBroker implements EventBroker { - constructor(logger: Logger); + // @deprecated + constructor(logger: LoggerService, events?: EventsService); // (undocumented) publish(params: EventParams): Promise; // (undocumented) @@ -23,7 +26,7 @@ export class DefaultEventBroker implements EventBroker { ): void; } -// @public +// @public @deprecated export class EventsBackend { constructor(logger: Logger); // (undocumented) diff --git a/plugins/events-backend/src/deprecated.ts b/plugins/events-backend/src/deprecated.ts new file mode 100644 index 0000000000..cce853b2af --- /dev/null +++ b/plugins/events-backend/src/deprecated.ts @@ -0,0 +1,18 @@ +/* + * 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 { EventsBackend } from './service/EventsBackend'; +export { DefaultEventBroker } from './service/DefaultEventBroker'; diff --git a/plugins/events-backend/src/index.ts b/plugins/events-backend/src/index.ts index be173b677c..63dfa5d252 100644 --- a/plugins/events-backend/src/index.ts +++ b/plugins/events-backend/src/index.ts @@ -20,6 +20,5 @@ * @packageDocumentation */ -export { EventsBackend } from './service/EventsBackend'; +export * from './deprecated'; export { HttpPostIngressEventPublisher } from './service/http'; -export { DefaultEventBroker } from './service/DefaultEventBroker'; diff --git a/plugins/events-backend/src/service/DefaultEventBroker.test.ts b/plugins/events-backend/src/service/DefaultEventBroker.test.ts index 99e5f6f72d..5317251726 100644 --- a/plugins/events-backend/src/service/DefaultEventBroker.test.ts +++ b/plugins/events-backend/src/service/DefaultEventBroker.test.ts @@ -85,15 +85,15 @@ describe('DefaultEventBroker', () => { } })(); - const errorSpy = jest.spyOn(logger, 'error'); + const warnSpy = jest.spyOn(logger, 'warn'); const eventBroker = new DefaultEventBroker(logger); eventBroker.subscribe(subscriber1); await eventBroker.publish({ topic, eventPayload: '1' }); - expect(errorSpy).toHaveBeenCalledTimes(1); - expect(errorSpy).toHaveBeenCalledWith( - 'Subscriber "Subscriber1" failed to process event', + expect(warnSpy).toHaveBeenCalledTimes(1); + expect(warnSpy).toHaveBeenCalledWith( + 'Subscriber "Subscriber1" failed to process event for topic "testTopic"', new Error('NOPE 1'), ); @@ -101,13 +101,13 @@ describe('DefaultEventBroker', () => { await eventBroker.publish({ topic, eventPayload: '2' }); // With two subscribers we should not halt on the first error but call all subscribers - expect(errorSpy).toHaveBeenCalledTimes(3); - expect(errorSpy).toHaveBeenCalledWith( - 'Subscriber "Subscriber1" failed to process event', + expect(warnSpy).toHaveBeenCalledTimes(3); + expect(warnSpy).toHaveBeenCalledWith( + 'Subscriber "Subscriber1" failed to process event for topic "testTopic"', new Error('NOPE 2'), ); - expect(errorSpy).toHaveBeenCalledWith( - 'Subscriber "Subscriber2" failed to process event', + expect(warnSpy).toHaveBeenCalledWith( + 'Subscriber "Subscriber2" failed to process event for topic "testTopic"', new Error('NOPE 2'), ); }); diff --git a/plugins/events-backend/src/service/DefaultEventBroker.ts b/plugins/events-backend/src/service/DefaultEventBroker.ts index c3824b3e7d..27523f3118 100644 --- a/plugins/events-backend/src/service/DefaultEventBroker.ts +++ b/plugins/events-backend/src/service/DefaultEventBroker.ts @@ -14,12 +14,14 @@ * limitations under the License. */ +import { LoggerService } from '@backstage/backend-plugin-api'; import { + DefaultEventsService, EventBroker, EventParams, + EventsService, EventSubscriber, } from '@backstage/plugin-events-node'; -import { Logger } from 'winston'; /** * In process event broker which will pass the event to all registered subscribers @@ -27,44 +29,34 @@ import { Logger } from 'winston'; * Events will not be persisted in any form. * * @public + * @deprecated use `DefaultEventsService` from `@backstage/plugin-events-node` instead */ -// TODO(pjungermann): add prom metrics? (see plugins/catalog-backend/src/util/metrics.ts, etc.) export class DefaultEventBroker implements EventBroker { - constructor(private readonly logger: Logger) {} + private readonly events: EventsService; - private readonly subscribers: { - [topic: string]: EventSubscriber[]; - } = {}; + /** + * + * @param logger - logger + * @param events - replacement that gets wrapped to support not yet migrated implementations. + * An instance can be passed (required for a mixed mode), otherwise a new instance gets created internally. + * @deprecated use `DefaultEventsService` directly instead + */ + constructor(logger: LoggerService, events?: EventsService) { + this.events = events ?? DefaultEventsService.create({ logger }); + } async publish(params: EventParams): Promise { - this.logger.debug( - `Event received: topic=${params.topic}, metadata=${JSON.stringify( - params.metadata, - )}, payload=${JSON.stringify(params.eventPayload)}`, - ); - - const subscribed = this.subscribers[params.topic] ?? []; - await Promise.all( - subscribed.map(async subscriber => { - try { - await subscriber.onEvent(params); - } catch (error) { - this.logger.error( - `Subscriber "${subscriber.constructor.name}" failed to process event`, - error, - ); - } - }), - ); + return this.events.publish(params); } subscribe( ...subscribers: Array> ): void { - subscribers.flat().forEach(subscriber => { - subscriber.supportsEventTopics().forEach(topic => { - this.subscribers[topic] = this.subscribers[topic] ?? []; - this.subscribers[topic].push(subscriber); + subscribers.flat().forEach(async subscriber => { + await this.events.subscribe({ + id: subscriber.constructor.name, + topics: subscriber.supportsEventTopics(), + onEvent: subscriber.onEvent.bind(subscriber), }); }); } diff --git a/plugins/events-backend/src/service/EventsBackend.ts b/plugins/events-backend/src/service/EventsBackend.ts index 4415b8703a..2c93663b46 100644 --- a/plugins/events-backend/src/service/EventsBackend.ts +++ b/plugins/events-backend/src/service/EventsBackend.ts @@ -26,6 +26,7 @@ import { DefaultEventBroker } from './DefaultEventBroker'; * A builder that helps wire up all component parts of the event management. * * @public + * @deprecated `EventBroker`, `EventPublisher`, and `EventSubscriber` got replaced by `EventsService` and its methods. */ export class EventsBackend { private eventBroker: EventBroker; diff --git a/plugins/events-node/api-report-alpha.md b/plugins/events-node/api-report-alpha.md index fd30f54d45..f61048ac94 100644 --- a/plugins/events-node/api-report-alpha.md +++ b/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> ): void; - // (undocumented) + // @deprecated (undocumented) addSubscribers( ...subscribers: Array> ): void; - // (undocumented) + // @deprecated (undocumented) setEventBroker(eventBroker: EventBroker): void; } diff --git a/plugins/events-node/api-report.md b/plugins/events-node/api-report.md index dfda48d97e..081c56549a 100644 --- a/plugins/events-node/api-report.md +++ b/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; + // (undocumented) + subscribe(options: EventsServiceSubscribeOptions): Promise; +} + +// @public @deprecated export interface EventBroker { publish(params: EventParams): Promise; subscribe( @@ -18,9 +32,9 @@ export interface EventParams { topic: string; } -// @public +// @public @deprecated export interface EventPublisher { - // (undocumented) + // @deprecated (undocumented) setEventBroker(eventBroker: EventBroker): Promise; } @@ -39,8 +53,29 @@ export abstract class EventRouter implements EventPublisher, EventSubscriber { } // @public +export interface EventsService { + publish(params: EventParams): Promise; + subscribe(options: EventsServiceSubscribeOptions): Promise; +} + +// @public (undocumented) +export type EventsServiceEventHandler = (params: EventParams) => Promise; + +// @public +export const eventsServiceRef: ServiceRef; + +// @public (undocumented) +export type EventsServiceSubscribeOptions = { + id: string; + topics: string[]; + onEvent: EventsServiceEventHandler; +}; + +// @public @deprecated export interface EventSubscriber { + // @deprecated onEvent(params: EventParams): Promise; + // @deprecated supportsEventTopics(): string[]; } diff --git a/plugins/events-node/package.json b/plugins/events-node/package.json index 58ddd50413..de794dfa45 100644 --- a/plugins/events-node/package.json +++ b/plugins/events-node/package.json @@ -45,6 +45,7 @@ "@backstage/backend-plugin-api": "workspace:^" }, "devDependencies": { + "@backstage/backend-common": "workspace:^", "@backstage/cli": "workspace:^" }, "files": [ diff --git a/plugins/events-node/src/api/DefaultEventsService.test.ts b/plugins/events-node/src/api/DefaultEventsService.test.ts new file mode 100644 index 0000000000..33df923892 --- /dev/null +++ b/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'), + ); + }); +}); diff --git a/plugins/events-node/src/api/DefaultEventsService.ts b/plugins/events-node/src/api/DefaultEventsService.ts new file mode 100644 index 0000000000..bb5c2a0ca8 --- /dev/null +++ b/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[] + >(); + + 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 => { + return this.publish(params); + }, + subscribe: (options: EventsServiceSubscribeOptions): Promise => { + return this.subscribe({ + ...options, + id: `${pluginId}.${options.id}`, + }); + }, + }; + } + + async publish(params: EventParams): Promise { + 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[] = []; + 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 { + options.topics.forEach(topic => { + if (!this.subscribers.has(topic)) { + this.subscribers.set(topic, []); + } + + this.subscribers.get(topic)!.push({ + id: options.id, + onEvent: options.onEvent, + }); + }); + } +} diff --git a/plugins/events-node/src/api/EventBroker.ts b/plugins/events-node/src/api/EventBroker.ts index 736c2a2bf0..f6afdf2f14 100644 --- a/plugins/events-node/src/api/EventBroker.ts +++ b/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 { /** diff --git a/plugins/events-node/src/api/EventPublisher.ts b/plugins/events-node/src/api/EventPublisher.ts index 285f427804..9089eb33fa 100644 --- a/plugins/events-node/src/api/EventPublisher.ts +++ b/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; } diff --git a/plugins/events-node/src/api/EventSubscriber.ts b/plugins/events-node/src/api/EventSubscriber.ts index 439f49b890..3686a8db3f 100644 --- a/plugins/events-node/src/api/EventSubscriber.ts +++ b/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; } diff --git a/plugins/events-node/src/api/EventsService.ts b/plugins/events-node/src/api/EventsService.ts new file mode 100644 index 0000000000..7af13f9b07 --- /dev/null +++ b/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; + + /** + * Subscribes to one or more topics, registering an event handler for them. + * + * @param options - event subscription options. + */ + subscribe(options: EventsServiceSubscribeOptions): Promise; +} + +/** + * @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; diff --git a/plugins/events-node/src/api/index.ts b/plugins/events-node/src/api/index.ts index 91711c0e38..94d3014dff 100644 --- a/plugins/events-node/src/api/index.ts +++ b/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'; diff --git a/plugins/events-node/src/deprecated.ts b/plugins/events-node/src/deprecated.ts new file mode 100644 index 0000000000..615e7b81ea --- /dev/null +++ b/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'; diff --git a/plugins/events-node/src/extensions.ts b/plugins/events-node/src/extensions.ts index 2e44b381af..90add52563 100644 --- a/plugins/events-node/src/extensions.ts +++ b/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> ): void; + /** + * @deprecated use `EventsService.subscribe` instead + */ addSubscribers( ...subscribers: Array> ): void; diff --git a/plugins/events-node/src/index.ts b/plugins/events-node/src/index.ts index 2bdf456f13..2bd93a8aea 100644 --- a/plugins/events-node/src/index.ts +++ b/plugins/events-node/src/index.ts @@ -21,3 +21,5 @@ */ export * from './api'; +export * from './deprecated'; +export { eventsServiceRef } from './service'; diff --git a/plugins/events-node/src/service.ts b/plugins/events-node/src/service.ts new file mode 100644 index 0000000000..e1d3047fca --- /dev/null +++ b/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({ + 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()); + }, + }), +}); diff --git a/yarn.lock b/yarn.lock index 7271c5f6bb..f49f9f73c5 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6495,6 +6495,7 @@ __metadata: version: 0.0.0-use.local resolution: "@backstage/plugin-events-node@workspace:plugins/events-node" dependencies: + "@backstage/backend-common": "workspace:^" "@backstage/backend-plugin-api": "workspace:^" "@backstage/cli": "workspace:^" languageName: unknown