diff --git a/plugins/events-backend/src/service/EventsPlugin.ts b/plugins/events-backend/src/service/EventsPlugin.ts index 089ea928bc..176d7e2f34 100644 --- a/plugins/events-backend/src/service/EventsPlugin.ts +++ b/plugins/events-backend/src/service/EventsPlugin.ts @@ -28,7 +28,7 @@ import { } from '@backstage/plugin-events-node'; import Router from 'express-promise-router'; import { HttpPostIngressEventPublisher } from './http'; -import { EventHub } from './hub'; +import { createEventBusRouter } from './hub'; class EventsExtensionPointImpl implements EventsExtensionPoint { #httpPostIngresses: HttpPostIngressOptions[] = []; @@ -97,8 +97,7 @@ export const eventsPlugin = createBackendPlugin({ const eventsRouter = Router(); http.bind(eventsRouter); - const hub = await EventHub.create({ database, logger, httpAuth }); - router.use(hub.handler()); + router.use(await createEventBusRouter({ database, logger, httpAuth })); router.use(eventsRouter); router.addAuthPolicy({ diff --git a/plugins/events-backend/src/service/hub/EventHub.ts b/plugins/events-backend/src/service/hub/EventHub.ts index 344f20970f..59be37967e 100644 --- a/plugins/events-backend/src/service/hub/EventHub.ts +++ b/plugins/events-backend/src/service/hub/EventHub.ts @@ -21,89 +21,45 @@ import { } from '@backstage/backend-plugin-api'; import { Handler } from 'express'; import Router from 'express-promise-router'; -import { spec, createOpenApiRouter } from '../../schema/openapi.generated'; -import { internal } from '@backstage/backend-openapi-utils'; +import { createOpenApiRouter } from '../../schema/openapi.generated'; import { MemoryEventHubStore } from './MemoryEventHubStore'; import { DatabaseEventHubStore } from './DatabaseEventHubStore'; import { EventHubStore } from './types'; import { EventParams } from '@backstage/plugin-events-node'; -export class EventHub { - static async create(options: { - logger: LoggerService; - database: DatabaseService; - httpAuth: HttpAuthService; - }) { - const { database, httpAuth } = options; - const logger = options.logger.child({ type: 'EventHub' }); - const router = Router(); +export async function createEventBusRouter(options: { + logger: LoggerService; + database: DatabaseService; + httpAuth: HttpAuthService; +}): Promise { + const { database, httpAuth } = options; + const logger = options.logger.child({ type: 'EventHub' }); + const router = Router(); - let store: EventHubStore; - const db = await database.getClient(); - if (db.client.config.client === 'pg') { - logger.info('Database is PostgreSQL, using database store'); - store = await DatabaseEventHubStore.create({ - database, - logger, - }); - } else { - logger.info('Database is not PostgreSQL, using memory store'); - store = new MemoryEventHubStore(); - } - - const hub = new EventHub(router, logger, httpAuth, store); - - const apiRouter = await createOpenApiRouter(); - - router.use(apiRouter); - - apiRouter.post('/hub/events', hub.#handlePostEvents); - - // Long-polling - apiRouter.get( - '/hub/subscriptions/:subscriptionId/events', - hub.#handleGetSubscription, - ); - apiRouter.put( - '/hub/subscriptions/:subscriptionId', - hub.#handlePutSubscription, - ); - - return hub; + let store: EventHubStore; + const db = await database.getClient(); + if (db.client.config.client === 'pg') { + logger.info('Database is PostgreSQL, using database store'); + store = await DatabaseEventHubStore.create({ + database, + logger, + }); + } else { + logger.info('Database is not PostgreSQL, using memory store'); + store = new MemoryEventHubStore(); } - readonly #handler: Handler; - readonly #logger: LoggerService; - readonly #httpAuth: HttpAuthService; - readonly #store: EventHubStore; + const apiRouter = await createOpenApiRouter(); - private constructor( - handler: Handler, - logger: LoggerService, - httpAuth: HttpAuthService, - store: EventHubStore, - ) { - this.#handler = handler; - this.#logger = logger; - this.#httpAuth = httpAuth; - this.#store = store; - } + router.use(apiRouter); - handler(): Handler { - return this.#handler; - } - - #handlePostEvents: internal.DocRequestHandler< - typeof spec, - '/hub/events', - 'post' - > = async (req, res) => { - const credentials = await this.#httpAuth.credentials(req, { + apiRouter.post('/hub/events', async (req, res) => { + const credentials = await httpAuth.credentials(req, { allow: ['service'], }); const topic = req.body.event.topic; const subscriberIds = req.body.subscriptionIds ?? []; - const result = await this.#store.publish({ + const result = await store.publish({ params: { topic, eventPayload: req.body.event.payload, @@ -111,15 +67,12 @@ export class EventHub { subscriberIds, }); if (result) { - this.#logger.info( - `Published event to '${topic}' with ID '${result.id}'`, - { - subject: credentials.principal.subject, - }, - ); + logger.info(`Published event to '${topic}' with ID '${result.id}'`, { + subject: credentials.principal.subject, + }); res.status(201).end(); } else { - this.#logger.info( + logger.info( `Skipped publishing of event to '${topic}', subscribers have already been notified: '${subscriberIds.join( "', '", )}'`, @@ -129,77 +82,74 @@ export class EventHub { ); res.status(204).end(); } - }; + }); - #handleGetSubscription: internal.DocRequestHandler< - typeof spec, - '/hub/subscriptions/{subscriptionId}/events', - 'get' - > = async (req, res) => { - const credentials = await this.#httpAuth.credentials(req, { - allow: ['service'], - }); - const id = req.params.subscriptionId; + apiRouter.get( + '/hub/subscriptions/:subscriptionId/events', + async (req, res) => { + const credentials = await httpAuth.credentials(req, { + allow: ['service'], + }); + const id = req.params.subscriptionId; - let resolveShouldNotify: (shouldNotify: boolean) => void; - const shouldNotifyPromise = new Promise(resolve => { - resolveShouldNotify = resolve; - }); + let resolveShouldNotify: (shouldNotify: boolean) => void; + const shouldNotifyPromise = new Promise(resolve => { + resolveShouldNotify = resolve; + }); - const { cancel } = await this.#store.listen(id, { - onNotify() { - shouldNotifyPromise.then(shouldNotify => { - if (shouldNotify) { - res.status(204).end(); - } - }); - }, - onError() { - shouldNotifyPromise.then(shouldNotify => { - if (shouldNotify) { - res.status(500).end(); - } - }); - }, - }); - req.on('end', cancel); + const { cancel } = await store.listen(id, { + onNotify() { + shouldNotifyPromise.then(shouldNotify => { + if (shouldNotify) { + res.status(204).end(); + } + }); + }, + onError() { + shouldNotifyPromise.then(shouldNotify => { + if (shouldNotify) { + res.status(500).end(); + } + }); + }, + }); + req.on('end', cancel); - try { - const { events } = await this.#store.readSubscription(id); + try { + const { events } = await store.readSubscription(id); - this.#logger.info( - `Reading subscription '${id}' resulted in ${events.length} events`, - { subject: credentials.principal.subject }, - ); + logger.info( + `Reading subscription '${id}' resulted in ${events.length} events`, + { subject: credentials.principal.subject }, + ); - if (events.length > 0) { - res.json({ events }); + if (events.length > 0) { + res.json({ events }); + resolveShouldNotify!(false); + } else { + resolveShouldNotify!(true); + } + } finally { resolveShouldNotify!(false); - } else { - resolveShouldNotify!(true); } - } finally { - resolveShouldNotify!(false); - } - }; + }, + ); - #handlePutSubscription: internal.DocRequestHandler< - typeof spec, - '/hub/subscriptions/{subscriptionId}', - 'put' - > = async (req, res) => { - const credentials = await this.#httpAuth.credentials(req, { + apiRouter.put('/hub/subscriptions/:subscriptionId', async (req, res) => { + const credentials = await httpAuth.credentials(req, { allow: ['service'], }); const id = req.params.subscriptionId; - await this.#store.upsertSubscription(id, req.body.topics); + await store.upsertSubscription(id, req.body.topics); - this.#logger.info( + logger.info( `New subscription '${id}' topics='${req.body.topics.join("', '")}'`, { subject: credentials.principal.subject }, ); res.status(201).end(); - }; + }); + + return apiRouter; } diff --git a/plugins/events-backend/src/service/hub/index.ts b/plugins/events-backend/src/service/hub/index.ts index 4acc848298..2e769f24d7 100644 --- a/plugins/events-backend/src/service/hub/index.ts +++ b/plugins/events-backend/src/service/hub/index.ts @@ -14,4 +14,4 @@ * limitations under the License. */ -export { EventHub } from './EventHub'; +export { createEventBusRouter } from './EventHub';