From f1fb54af5b605e9db0fc2440fb1e220e41aae598 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Fri, 13 Sep 2024 09:29:15 +0200 Subject: [PATCH] events-backend: rename consumedBy -> notifiedSubscribers Signed-off-by: Patrik Oldsberg --- .../src/schema/openapi.generated.ts | 2 +- plugins/events-backend/src/schema/openapi.yaml | 2 +- .../src/service/EventsPlugin.test.ts | 17 ++++++++++++----- .../src/service/hub/DatabaseEventBusStore.ts | 8 ++++---- .../src/service/hub/MemoryEventBusStore.ts | 14 ++++++++------ .../src/service/hub/createEventBusRouter.ts | 8 ++++---- plugins/events-backend/src/service/hub/types.ts | 2 +- .../events-node/src/api/DefaultEventsService.ts | 2 +- .../generated/models/PostEventRequest.model.ts | 2 +- 9 files changed, 33 insertions(+), 24 deletions(-) diff --git a/plugins/events-backend/src/schema/openapi.generated.ts b/plugins/events-backend/src/schema/openapi.generated.ts index ef712e2f86..8f8d6c6a82 100644 --- a/plugins/events-backend/src/schema/openapi.generated.ts +++ b/plugins/events-backend/src/schema/openapi.generated.ts @@ -159,7 +159,7 @@ export const spec = { event: { $ref: '#/components/schemas/Event', }, - consumedBy: { + notifiedSubscribers: { type: 'array', description: 'The IDs of subscriptions that have already received this event', diff --git a/plugins/events-backend/src/schema/openapi.yaml b/plugins/events-backend/src/schema/openapi.yaml index 8b2639a189..97f7705f87 100644 --- a/plugins/events-backend/src/schema/openapi.yaml +++ b/plugins/events-backend/src/schema/openapi.yaml @@ -106,7 +106,7 @@ paths: properties: event: $ref: '#/components/schemas/Event' - consumedBy: + notifiedSubscribers: type: array description: The IDs of subscriptions that have already received this event items: diff --git a/plugins/events-backend/src/service/EventsPlugin.test.ts b/plugins/events-backend/src/service/EventsPlugin.test.ts index ba9a236800..dcf901b1a0 100644 --- a/plugins/events-backend/src/service/EventsPlugin.test.ts +++ b/plugins/events-backend/src/service/EventsPlugin.test.ts @@ -119,12 +119,15 @@ describe('eventsPlugin', () => { publish( topic: string, payload: unknown, - options?: { consumedBy?: string[] }, + options?: { notifiedSubscribers?: string[] }, ) { return request(this.backend.server) .post('/api/events/bus/v1/events') .set('authorization', mockCredentials.service.header()) - .send({ event: { topic, payload }, consumedBy: options?.consumedBy }); + .send({ + event: { topic, payload }, + notifiedSubscribers: options?.notifiedSubscribers, + }); } readEvents(id: string) { @@ -278,7 +281,7 @@ describe('eventsPlugin', () => { 'test', { for: 'tester-2' }, { - consumedBy: ['tester-1'], + notifiedSubscribers: ['tester-1'], }, ) .expect(201); @@ -287,7 +290,7 @@ describe('eventsPlugin', () => { 'test', { for: 'tester-1' }, { - consumedBy: ['tester-2'], + notifiedSubscribers: ['tester-2'], }, ) .expect(201); @@ -362,7 +365,11 @@ describe('eventsPlugin', () => { await helper.subscribe('tester', ['test']).expect(201); await helper - .publish('test', { for: 'tester-2' }, { consumedBy: ['tester'] }) + .publish( + 'test', + { for: 'tester-2' }, + { notifiedSubscribers: ['tester'] }, + ) .expect(204); await backend.stop(); diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts index b8a3e54ab8..a41e95733d 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts @@ -349,10 +349,10 @@ export class DatabaseEventBusStore implements EventBusStore { async publish(options: { params: EventParams; - consumedBy?: string[]; + notifiedSubscribers?: string[]; }): Promise<{ eventId: string } | undefined> { const topic = options.params.topic; - const consumedBy = options.consumedBy ?? []; + const notifiedSubscribers = options.notifiedSubscribers ?? []; // This query inserts a new event into the database, but only if there are // subscribers to the topic that have not already been notified const result = await this.#db @@ -378,12 +378,12 @@ export class DatabaseEventBusStore implements EventBusStore { metadata: options.params.metadata, }), ]), - this.#db.raw('?', [consumedBy]), + this.#db.raw('?', [notifiedSubscribers]), ) // The rest of this query is to check whether there are any // subscribers that have not been notified yet .from(TABLE_SUBSCRIPTIONS) - .whereNotIn('id', consumedBy) // Skip notified subscribers + .whereNotIn('id', notifiedSubscribers) // Skip notified subscribers .andWhere(this.#db.raw('? = ANY(topics)', [topic])) // Match topic .having(this.#db.raw('count(*)'), '>', 0), // Check if there are any results ) diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts index 1527eeb1fa..6c7b4e6df7 100644 --- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts @@ -20,7 +20,9 @@ import { NotFoundError } from '@backstage/errors'; const MAX_BATCH_SIZE = 10; export class MemoryEventBusStore implements EventBusStore { - #events = new Array }>(); + #events = new Array< + EventParams & { seq: number; notifiedSubscribers: Set } + >(); #subscribers = new Map< string, { id: string; seq: number; topics: Set } @@ -32,14 +34,14 @@ export class MemoryEventBusStore implements EventBusStore { async publish(options: { params: EventParams; - consumedBy: string[]; + notifiedSubscribers: string[]; }): Promise<{ eventId: string } | undefined> { const topic = options.params.topic; - const consumedBy = new Set(options.consumedBy); + const notifiedSubscribers = new Set(options.notifiedSubscribers); let hasOtherSubscribers = false; for (const sub of this.#subscribers.values()) { - if (sub.topics.has(topic) && !consumedBy.has(sub.id)) { + if (sub.topics.has(topic) && !notifiedSubscribers.has(sub.id)) { hasOtherSubscribers = true; break; } @@ -49,7 +51,7 @@ export class MemoryEventBusStore implements EventBusStore { } const nextSeq = this.#getMaxSeq() + 1; - this.#events.push({ ...options.params, consumedBy, seq: nextSeq }); + this.#events.push({ ...options.params, notifiedSubscribers, seq: nextSeq }); for (const listener of this.#listeners) { if (listener.topics.has(topic)) { @@ -88,7 +90,7 @@ export class MemoryEventBusStore implements EventBusStore { event => event.seq > sub.seq && sub.topics.has(event.topic) && - !event.consumedBy.has(id), + !event.notifiedSubscribers.has(id), ) .slice(0, MAX_BATCH_SIZE); diff --git a/plugins/events-backend/src/service/hub/createEventBusRouter.ts b/plugins/events-backend/src/service/hub/createEventBusRouter.ts index fd0732e279..193c1108d5 100644 --- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts +++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts @@ -158,13 +158,13 @@ export async function createEventBusRouter(options: { allow: ['service'], }); const topic = req.body.event.topic; - const consumedBy = req.body.consumedBy; + const notifiedSubscribers = req.body.notifiedSubscribers; const result = await store.publish({ params: { topic, eventPayload: req.body.event.payload, } as EventParams, - consumedBy: req.body.consumedBy, + notifiedSubscribers: req.body.notifiedSubscribers, }); if (result) { logger.info(`Published event to '${topic}' with ID '${result.eventId}'`, { @@ -172,8 +172,8 @@ export async function createEventBusRouter(options: { }); res.status(201).end(); } else { - if (consumedBy) { - const notified = `'${consumedBy.join("', '")}'`; + if (notifiedSubscribers) { + const notified = `'${notifiedSubscribers.join("', '")}'`; logger.info( `Skipped publishing of event to '${topic}', subscribers have already been notified: ${notified}`, { subject: credentials.principal.subject }, diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts index b221f6696d..3a4c8df9f8 100644 --- a/plugins/events-backend/src/service/hub/types.ts +++ b/plugins/events-backend/src/service/hub/types.ts @@ -19,7 +19,7 @@ import { EventParams } from '@backstage/plugin-events-node'; export type EventBusStore = { publish(options: { params: EventParams; - consumedBy?: string[]; + notifiedSubscribers?: string[]; }): Promise<{ eventId: string } | undefined>; upsertSubscription(subscriptionId: string, topics: string[]): Promise; diff --git a/plugins/events-node/src/api/DefaultEventsService.ts b/plugins/events-node/src/api/DefaultEventsService.ts index 9e2509539a..9055afaeb8 100644 --- a/plugins/events-node/src/api/DefaultEventsService.ts +++ b/plugins/events-node/src/api/DefaultEventsService.ts @@ -127,7 +127,7 @@ class PluginEventsService implements EventsService { { body: { event: { payload: params.eventPayload, topic: params.topic }, - consumedBy: notifiedSubscribers, + notifiedSubscribers, }, }, { token }, diff --git a/plugins/events-node/src/generated/models/PostEventRequest.model.ts b/plugins/events-node/src/generated/models/PostEventRequest.model.ts index 04eaaab461..75289ed25b 100644 --- a/plugins/events-node/src/generated/models/PostEventRequest.model.ts +++ b/plugins/events-node/src/generated/models/PostEventRequest.model.ts @@ -24,5 +24,5 @@ export interface PostEventRequest { /** * The IDs of subscriptions that have already received this event */ - consumedBy?: Array; + notifiedSubscribers?: Array; }