From 12dab612507fd08fa42c318158f5d1785b4196a2 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Mon, 27 May 2024 09:47:28 +0200 Subject: [PATCH] events-backend: use consumedBy everywhere instead of subscriptionIds Signed-off-by: Patrik Oldsberg --- .../src/schema/openapi.generated.ts | 2 +- .../events-backend/src/schema/openapi.yaml | 2 +- .../src/service/hub/DatabaseEventBusStore.ts | 7 ++--- .../src/service/hub/MemoryEventBusStore.ts | 14 +++++----- .../src/service/hub/createEventBusRouter.ts | 26 +++++++++++-------- .../events-backend/src/service/hub/types.ts | 2 +- .../src/api/DefaultEventsService.ts | 2 +- .../models/PostEventRequest.model.ts | 2 +- 8 files changed, 30 insertions(+), 27 deletions(-) diff --git a/plugins/events-backend/src/schema/openapi.generated.ts b/plugins/events-backend/src/schema/openapi.generated.ts index d6278ff070..ef712e2f86 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', }, - subscriptionIds: { + consumedBy: { 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 48f2c74e5e..8b2639a189 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' - subscriptionIds: + consumedBy: type: array description: The IDs of subscriptions that have already received this event items: diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts index 0a8dddfe51..87c9263e79 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts @@ -323,9 +323,10 @@ export class DatabaseEventBusStore implements EventBusStore { async publish(options: { params: EventParams; - subscriberIds: string[]; + consumedBy?: string[]; }): Promise<{ id: string } | undefined> { const topic = options.params.topic; + const consumedBy = options.consumedBy ?? []; // 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 @@ -351,12 +352,12 @@ export class DatabaseEventBusStore implements EventBusStore { metadata: options.params.metadata, }), ]), - this.#db.raw('?', [options.subscriberIds]), + this.#db.raw('?', [consumedBy]), ) // The rest of this query is to check whether there are any // subscribers that have not been notified yet .from(TABLE_SUBSCRIPTIONS) - .whereNotIn('id', options.subscriberIds) // Skip notified subscribers + .whereNotIn('id', consumedBy) // 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 0fc9a2b14f..23aec6ba97 100644 --- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts @@ -19,9 +19,7 @@ import { EventBusStore } from './types'; const MAX_BATCH_SIZE = 5; export class MemoryEventBusStore implements EventBusStore { - #events = new Array< - EventParams & { seq: number; subscriberIds: Set } - >(); + #events = new Array }>(); #subscribers = new Map< string, { id: string; seq: number; topics: Set } @@ -33,14 +31,14 @@ export class MemoryEventBusStore implements EventBusStore { async publish(options: { params: EventParams; - subscriberIds: string[]; + consumedBy: string[]; }): Promise<{ id: string } | undefined> { const topic = options.params.topic; - const subscriberIds = new Set(options.subscriberIds); + const consumedBy = new Set(options.consumedBy); let hasOtherSubscribers = false; for (const sub of this.#subscribers.values()) { - if (sub.topics.has(topic) && !subscriberIds.has(sub.id)) { + if (sub.topics.has(topic) && !consumedBy.has(sub.id)) { hasOtherSubscribers = true; break; } @@ -50,7 +48,7 @@ export class MemoryEventBusStore implements EventBusStore { } const nextSeq = this.#getMaxSeq() + 1; - this.#events.push({ ...options.params, subscriberIds, seq: nextSeq }); + this.#events.push({ ...options.params, consumedBy, seq: nextSeq }); for (const listener of this.#listeners) { if (listener.topics.has(topic)) { @@ -89,7 +87,7 @@ export class MemoryEventBusStore implements EventBusStore { event => event.seq > sub.seq && sub.topics.has(event.topic) && - !event.subscriberIds.has(id), + !event.consumedBy.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 1a702c624b..566574f6e8 100644 --- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts +++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts @@ -160,13 +160,13 @@ export async function createEventBusRouter(options: { allow: ['service'], }); const topic = req.body.event.topic; - const subscriberIds = req.body.subscriptionIds ?? []; + const consumedBy = req.body.consumedBy; const result = await store.publish({ params: { topic, eventPayload: req.body.event.payload, } as EventParams, - subscriberIds, + consumedBy: req.body.consumedBy, }); if (result) { logger.info(`Published event to '${topic}' with ID '${result.id}'`, { @@ -174,14 +174,18 @@ export async function createEventBusRouter(options: { }); res.status(201).end(); } else { - logger.info( - `Skipped publishing of event to '${topic}', subscribers have already been notified: '${subscriberIds.join( - "', '", - )}'`, - { - subject: credentials.principal.subject, - }, - ); + if (consumedBy) { + const notified = `'${consumedBy.join("', '")}'`; + logger.info( + `Skipped publishing of event to '${topic}', subscribers have already been notified: ${notified}`, + { subject: credentials.principal.subject }, + ); + } else { + logger.info( + `Skipped publishing of event to '${topic}', no subscribers present`, + { subject: credentials.principal.subject }, + ); + } res.status(204).end(); } }); @@ -267,7 +271,7 @@ export async function createEventBusRouter(options: { await store.upsertSubscription(id, req.body.topics); logger.info( - `New subscription '${id}' topics='${req.body.topics.join("', '")}'`, + `New subscription '${id}' for topics '${req.body.topics.join("', '")}'`, { 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 87e62049a9..315a4a7220 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; - subscriberIds: string[]; + consumedBy?: string[]; }): Promise<{ id: string } | undefined>; upsertSubscription(id: string, topics: string[]): Promise; diff --git a/plugins/events-node/src/api/DefaultEventsService.ts b/plugins/events-node/src/api/DefaultEventsService.ts index 74e37100e0..5a2712fc89 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 }, - subscriptionIds: notifiedSubscribers, + consumedBy: 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 527064bc13..04eaaab461 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 */ - subscriptionIds?: Array; + consumedBy?: Array; }