From c4bd79422ab24f92bf706a79ccc1fb11446ff932 Mon Sep 17 00:00:00 2001 From: Patrick Jungermann Date: Tue, 23 Jan 2024 20:28:35 +0100 Subject: [PATCH] feat(events)!: migrate `HttpPostIngressEventPublisher` and `eventsPlugin` to use `EventsService` Signed-off-by: Patrick Jungermann --- .changeset/young-flies-wash.md | 35 +++++++++++ packages/backend/src/index.ts | 8 ++- packages/backend/src/plugins/events.ts | 11 +--- packages/backend/src/types.ts | 6 +- plugins/events-backend/api-report.md | 7 +-- .../src/service/EventsPlugin.test.ts | 56 +++++++++++------- .../src/service/EventsPlugin.ts | 59 ++++++------------- .../HttpPostIngressEventPublisher.test.ts | 40 ++++++------- .../http/HttpPostIngressEventPublisher.ts | 23 +++----- 9 files changed, 132 insertions(+), 113 deletions(-) create mode 100644 .changeset/young-flies-wash.md diff --git a/.changeset/young-flies-wash.md b/.changeset/young-flies-wash.md new file mode 100644 index 0000000000..dcedc9bb7c --- /dev/null +++ b/.changeset/young-flies-wash.md @@ -0,0 +1,35 @@ +--- +'@backstage/plugin-events-backend': minor +--- + +BREAKING CHANGE: Migrate `HttpPostIngressEventPublisher` and `eventsPlugin` to use `EventsService`. + +Uses the `EventsService` instead of `EventBroker` at `HttpPostIngressEventPublisher`, +dropping the use of `EventPublisher` including `setEventBroker(..)`. + +Now, `HttpPostIngressEventPublisher.fromConfig` requires `events: EventsService` as option. + +```diff + const http = HttpPostIngressEventPublisher.fromConfig({ + config: env.config, ++ events: env.events, + logger: env.logger, + }); + http.bind(eventsRouter); + + // e.g. at packages/backend/src/plugins/events.ts +- await new EventsBackend(env.logger) +- .setEventBroker(env.eventBroker) +- .addPublishers(http) +- .start(); + + // or for other kinds of setups +- await Promise.all(http.map(publisher => publisher.setEventBroker(eventBroker))); +``` + +`eventsPlugin` uses the `eventsServiceRef` as dependency. +Unsupported (and deprecated) extension point methods will throw an error to prevent unintended behavior. + +```ts +import { eventsServiceRef } from '@backstage/plugin-events-node'; +``` diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index bade028597..8931e6832b 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -70,6 +70,7 @@ import { PluginEnvironment } from './types'; import { ServerPermissionClient } from '@backstage/plugin-permission-node'; import { DefaultIdentityClient } from '@backstage/plugin-auth-node'; import { DefaultEventBroker } from '@backstage/plugin-events-backend'; +import { DefaultEventsService } from '@backstage/plugin-events-node'; import { PrometheusExporter } from '@opentelemetry/exporter-prometheus'; import { MeterProvider } from '@opentelemetry/sdk-metrics'; import { metrics } from '@opentelemetry/api'; @@ -99,7 +100,11 @@ function makeCreateEnv(config: Config) { discovery, }); - const eventBroker = new DefaultEventBroker(root.child({ type: 'plugin' })); + const eventsService = DefaultEventsService.create({ logger: root }); + const eventBroker = new DefaultEventBroker( + root.child({ type: 'plugin' }), + eventsService, + ); const signalService = DefaultSignalService.create({ eventBroker, }); @@ -119,6 +124,7 @@ function makeCreateEnv(config: Config) { config, reader, eventBroker, + events: eventsService, discovery, tokenManager, permissions, diff --git a/packages/backend/src/plugins/events.ts b/packages/backend/src/plugins/events.ts index f3ff354240..fd60a9bb14 100644 --- a/packages/backend/src/plugins/events.ts +++ b/packages/backend/src/plugins/events.ts @@ -14,10 +14,7 @@ * limitations under the License. */ -import { - EventsBackend, - HttpPostIngressEventPublisher, -} from '@backstage/plugin-events-backend'; +import { HttpPostIngressEventPublisher } from '@backstage/plugin-events-backend'; import { Router } from 'express'; import { PluginEnvironment } from '../types'; @@ -28,14 +25,10 @@ export default async function createPlugin( const http = HttpPostIngressEventPublisher.fromConfig({ config: env.config, + events: env.events, logger: env.logger, }); http.bind(eventsRouter); - await new EventsBackend(env.logger) - .setEventBroker(env.eventBroker) - .addPublishers(http) - .start(); - return eventsRouter; } diff --git a/packages/backend/src/types.ts b/packages/backend/src/types.ts index d76e68c1c9..7d9cd19310 100644 --- a/packages/backend/src/types.ts +++ b/packages/backend/src/types.ts @@ -26,7 +26,7 @@ import { import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { IdentityApi } from '@backstage/plugin-auth-node'; import { PermissionEvaluator } from '@backstage/plugin-permission-common'; -import { EventBroker } from '@backstage/plugin-events-node'; +import { EventBroker, EventsService } from '@backstage/plugin-events-node'; import { SignalService } from '@backstage/plugin-signals-node'; export type PluginEnvironment = { @@ -40,6 +40,10 @@ export type PluginEnvironment = { permissions: PermissionEvaluator; scheduler: PluginTaskScheduler; identity: IdentityApi; + /** + * @deprecated use `events` instead + */ eventBroker: EventBroker; + events: EventsService; signalService: SignalService; }; diff --git a/plugins/events-backend/api-report.md b/plugins/events-backend/api-report.md index 9fffe719ba..8b5d4e8d31 100644 --- a/plugins/events-backend/api-report.md +++ b/plugins/events-backend/api-report.md @@ -43,18 +43,17 @@ export class EventsBackend { } // @public -export class HttpPostIngressEventPublisher implements EventPublisher { +export class HttpPostIngressEventPublisher { // (undocumented) bind(router: express.Router): void; // (undocumented) static fromConfig(env: { config: Config; + events: EventsService; ingresses?: { [topic: string]: Omit; }; - logger: Logger; + logger: LoggerService; }): HttpPostIngressEventPublisher; - // (undocumented) - setEventBroker(eventBroker: EventBroker): Promise; } ``` diff --git a/plugins/events-backend/src/service/EventsPlugin.test.ts b/plugins/events-backend/src/service/EventsPlugin.test.ts index 7f555a0b6f..e39951e8dc 100644 --- a/plugins/events-backend/src/service/EventsPlugin.test.ts +++ b/plugins/events-backend/src/service/EventsPlugin.test.ts @@ -14,22 +14,27 @@ * limitations under the License. */ -import { createBackendModule } from '@backstage/backend-plugin-api'; -import { mockServices, startTestBackend } from '@backstage/backend-test-utils'; import { - TestEventBroker, - TestEventPublisher, - TestEventSubscriber, -} from '@backstage/plugin-events-backend-test-utils'; + createBackendModule, + createServiceFactory, +} from '@backstage/backend-plugin-api'; +import { mockServices, startTestBackend } from '@backstage/backend-test-utils'; +import { eventsServiceRef } from '@backstage/plugin-events-node'; import { eventsExtensionPoint } from '@backstage/plugin-events-node/alpha'; +import { TestEventsService } from '@backstage/plugin-events-backend-test-utils'; import request from 'supertest'; import { eventsPlugin } from './EventsPlugin'; -describe('eventPlugin', () => { +describe('eventsPlugin', () => { it('should be initialized properly', async () => { - const eventBroker = new TestEventBroker(); - const publisher = new TestEventPublisher(); - const subscriber = new TestEventSubscriber('sub', ['fake']); + const eventsService = new TestEventsService(); + const eventsServiceFactory = createServiceFactory({ + service: eventsServiceRef, + deps: {}, + async factory({}) { + return eventsService; + }, + }); const testModule = createBackendModule({ pluginId: 'events', @@ -40,9 +45,9 @@ describe('eventPlugin', () => { events: eventsExtensionPoint, }, async init({ events }) { - events.setEventBroker(eventBroker); - events.addPublishers(publisher); - events.addSubscribers(subscriber); + events.addHttpPostIngress({ + topic: 'fake-ext', + }); }, }); }, @@ -51,6 +56,7 @@ describe('eventPlugin', () => { const { server } = await startTestBackend({ extensionPoints: [], features: [ + eventsServiceFactory(), eventsPlugin(), testModule(), mockServices.logger.factory(), @@ -66,18 +72,24 @@ describe('eventPlugin', () => { ], }); - expect(publisher.eventBroker).toBe(eventBroker); - expect(eventBroker.subscribed.length).toEqual(1); - expect(eventBroker.subscribed[0]).toBe(subscriber); - - const response = await request(server) + const response1 = await request(server) .post('/api/events/http/fake') .timeout(1000) .send({ test: 'fake' }); - expect(response.status).toBe(202); + expect(response1.status).toBe(202); - expect(eventBroker.published.length).toEqual(1); - expect(eventBroker.published[0].topic).toEqual('fake'); - expect(eventBroker.published[0].eventPayload).toEqual({ test: 'fake' }); + const response2 = await request(server) + .post('/api/events/http/fake-ext') + .timeout(1000) + .send({ test: 'fake-ext' }); + expect(response2.status).toBe(202); + + expect(eventsService.published).toHaveLength(2); + expect(eventsService.published[0].topic).toEqual('fake'); + expect(eventsService.published[0].eventPayload).toEqual({ test: 'fake' }); + expect(eventsService.published[1].topic).toEqual('fake-ext'); + expect(eventsService.published[1].eventPayload).toEqual({ + test: 'fake-ext', + }); }); }); diff --git a/plugins/events-backend/src/service/EventsPlugin.ts b/plugins/events-backend/src/service/EventsPlugin.ts index 27456b5b2a..5e1df975c5 100644 --- a/plugins/events-backend/src/service/EventsPlugin.ts +++ b/plugins/events-backend/src/service/EventsPlugin.ts @@ -18,59 +18,42 @@ import { createBackendPlugin, coreServices, } from '@backstage/backend-plugin-api'; -import { loggerToWinstonLogger } from '@backstage/backend-common'; import { eventsExtensionPoint, EventsExtensionPoint, } from '@backstage/plugin-events-node/alpha'; import { - EventBroker, - EventPublisher, - EventSubscriber, + eventsServiceRef, HttpPostIngressOptions, } from '@backstage/plugin-events-node'; -import { DefaultEventBroker } from './DefaultEventBroker'; import Router from 'express-promise-router'; import { HttpPostIngressEventPublisher } from './http'; class EventsExtensionPointImpl implements EventsExtensionPoint { - #eventBroker: EventBroker | undefined; #httpPostIngresses: HttpPostIngressOptions[] = []; - #publishers: EventPublisher[] = []; - #subscribers: EventSubscriber[] = []; - setEventBroker(eventBroker: EventBroker): void { - this.#eventBroker = eventBroker; + setEventBroker(_: any): void { + throw new Error( + 'setEventBroker is not supported anymore; use eventsServiceRef instead', + ); } - addPublishers( - ...publishers: Array> - ): void { - this.#publishers.push(...publishers.flat()); + addPublishers(_: any): void { + throw new Error( + 'addPublishers is not supported anymore; use EventsService instead', + ); } - addSubscribers( - ...subscribers: Array> - ): void { - this.#subscribers.push(...subscribers.flat()); + addSubscribers(_: any): void { + throw new Error( + 'addSubscribers is not supported anymore; use EventsService instead', + ); } addHttpPostIngress(options: HttpPostIngressOptions) { this.#httpPostIngresses.push(options); } - get eventBroker() { - return this.#eventBroker; - } - - get publishers() { - return this.#publishers; - } - - get subscribers() { - return this.#subscribers; - } - get httpPostIngresses() { return this.#httpPostIngresses; } @@ -90,12 +73,11 @@ export const eventsPlugin = createBackendPlugin({ env.registerInit({ deps: { config: coreServices.rootConfig, + events: eventsServiceRef, logger: coreServices.logger, router: coreServices.httpRouter, }, - async init({ config, logger, router }) { - const winstonLogger = loggerToWinstonLogger(logger); - + async init({ config, events, logger, router }) { const ingresses = Object.fromEntries( extensionPoint.httpPostIngresses.map(ingress => [ ingress.topic, @@ -105,20 +87,13 @@ export const eventsPlugin = createBackendPlugin({ const http = HttpPostIngressEventPublisher.fromConfig({ config, + events, ingresses, - logger: winstonLogger, + logger, }); const eventsRouter = Router(); http.bind(eventsRouter); router.use(eventsRouter); - - const eventBroker = - extensionPoint.eventBroker ?? new DefaultEventBroker(winstonLogger); - - eventBroker.subscribe(extensionPoint.subscribers); - [extensionPoint.publishers, http] - .flat() - .forEach(publisher => publisher.setEventBroker(eventBroker)); }, }); }, diff --git a/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.test.ts b/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.test.ts index 72e0b94c57..665ac0b4a9 100644 --- a/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.test.ts +++ b/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.test.ts @@ -16,7 +16,7 @@ import { getVoidLogger } from '@backstage/backend-common'; import { ConfigReader } from '@backstage/config'; -import { TestEventBroker } from '@backstage/plugin-events-backend-test-utils'; +import { TestEventsService } from '@backstage/plugin-events-backend-test-utils'; import express from 'express'; import Router from 'express-promise-router'; import request from 'supertest'; @@ -36,9 +36,11 @@ describe('HttpPostIngressEventPublisher', () => { const router = Router(); const app = express().use(router); + const events = new TestEventsService(); const publisher = HttpPostIngressEventPublisher.fromConfig({ config, + events, ingresses: { testB: {}, }, @@ -46,9 +48,6 @@ describe('HttpPostIngressEventPublisher', () => { }); publisher.bind(router); - const eventBroker = new TestEventBroker(); - await publisher.setEventBroker(eventBroker); - const notFoundResponse = await request(app) .post('/http/unknown') .timeout(1000) @@ -69,18 +68,18 @@ describe('HttpPostIngressEventPublisher', () => { .send({ testB: 'data' }); expect(response2.status).toBe(202); - expect(eventBroker.published.length).toEqual(2); - expect(eventBroker.published[0].topic).toEqual('testA'); - expect(eventBroker.published[0].eventPayload).toEqual({ testA: 'data' }); - expect(eventBroker.published[0].metadata).toEqual( + expect(events.published).toHaveLength(2); + expect(events.published[0].topic).toEqual('testA'); + expect(events.published[0].eventPayload).toEqual({ testA: 'data' }); + expect(events.published[0].metadata).toEqual( expect.objectContaining({ 'content-type': 'application/json', 'x-custom-header': 'test-value', }), ); - expect(eventBroker.published[1].topic).toEqual('testB'); - expect(eventBroker.published[1].eventPayload).toEqual({ testB: 'data' }); - expect(eventBroker.published[1].metadata).toEqual( + expect(events.published[1].topic).toEqual('testB'); + expect(events.published[1].eventPayload).toEqual({ testB: 'data' }); + expect(events.published[1].metadata).toEqual( expect.objectContaining({ 'content-type': 'application/json', 'x-custom-header': 'test-value', @@ -99,9 +98,11 @@ describe('HttpPostIngressEventPublisher', () => { const router = Router(); const app = express().use(router); + const events = new TestEventsService(); const publisher = HttpPostIngressEventPublisher.fromConfig({ config, + events, ingresses: { testB: { validator: async (req, context) => { @@ -146,9 +147,6 @@ describe('HttpPostIngressEventPublisher', () => { }); publisher.bind(router); - const eventBroker = new TestEventBroker(); - await publisher.setEventBroker(eventBroker); - const response1 = await request(app) .post('/http/testA') .timeout(1000) @@ -191,12 +189,12 @@ describe('HttpPostIngressEventPublisher', () => { expect(response6.status).toBe(403); expect(response6.body).toEqual({}); - expect(eventBroker.published.length).toEqual(2); - expect(eventBroker.published[0].topic).toEqual('testA'); - expect(eventBroker.published[0].eventPayload).toEqual({ test: 'data' }); - expect(eventBroker.published[1].topic).toEqual('testB'); - expect(eventBroker.published[1].eventPayload).toEqual({ test: 'data' }); - expect(eventBroker.published[1].metadata).toEqual( + expect(events.published).toHaveLength(2); + expect(events.published[0].topic).toEqual('testA'); + expect(events.published[0].eventPayload).toEqual({ test: 'data' }); + expect(events.published[1].topic).toEqual('testB'); + expect(events.published[1].eventPayload).toEqual({ test: 'data' }); + expect(events.published[1].metadata).toEqual( expect.objectContaining({ 'x-test-signature': 'testB-signature', }), @@ -205,10 +203,12 @@ describe('HttpPostIngressEventPublisher', () => { it('without configuration', async () => { const config = new ConfigReader({}); + const events = new TestEventsService(); expect(() => HttpPostIngressEventPublisher.fromConfig({ config, + events, logger, }), ).not.toThrow(); diff --git a/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts b/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts index b5a6ccbca1..06dc4e463a 100644 --- a/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts +++ b/plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts @@ -15,16 +15,15 @@ */ import { errorHandler } from '@backstage/backend-common'; +import { LoggerService } from '@backstage/backend-plugin-api'; import { Config } from '@backstage/config'; import { - EventBroker, - EventPublisher, + EventsService, HttpPostIngressOptions, RequestValidator, } from '@backstage/plugin-events-node'; import express from 'express'; import Router from 'express-promise-router'; -import { Logger } from 'winston'; import { RequestValidationContextImpl } from './validation'; /** @@ -34,13 +33,12 @@ import { RequestValidationContextImpl } from './validation'; * @public */ // TODO(pjungermann): add prom metrics? (see plugins/catalog-backend/src/util/metrics.ts, etc.) -export class HttpPostIngressEventPublisher implements EventPublisher { - private eventBroker?: EventBroker; - +export class HttpPostIngressEventPublisher { static fromConfig(env: { config: Config; + events: EventsService; ingresses?: { [topic: string]: Omit }; - logger: Logger; + logger: LoggerService; }): HttpPostIngressEventPublisher { const topics = env.config.getOptionalStringArray('events.http.topics') ?? []; @@ -54,11 +52,12 @@ export class HttpPostIngressEventPublisher implements EventPublisher { } }); - return new HttpPostIngressEventPublisher(env.logger, ingresses); + return new HttpPostIngressEventPublisher(env.events, env.logger, ingresses); } private constructor( - private readonly logger: Logger, + private readonly events: EventsService, + private readonly logger: LoggerService, private readonly ingresses: { [topic: string]: Omit; }, @@ -68,10 +67,6 @@ export class HttpPostIngressEventPublisher implements EventPublisher { router.use('/http', this.createRouter(this.ingresses)); } - async setEventBroker(eventBroker: EventBroker): Promise { - this.eventBroker = eventBroker; - } - private createRouter(ingresses: { [topic: string]: Omit; }): express.Router { @@ -108,7 +103,7 @@ export class HttpPostIngressEventPublisher implements EventPublisher { } const eventPayload = request.body; - await this.eventBroker!.publish({ + await this.events.publish({ topic, eventPayload, metadata: request.headers,