paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 51 Packages Projects Releases Wiki Activity

events-backend: couple of simple review fixes

Browse Source 
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2024-09-12 22:18:57 +02:00
parent b9daa73fed
commit f06211e447
4 changed files with 41 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/events-backend/migrations/20240523100528_init.js
+2 -1
View File
@@ -82,6 +82,7 @@ exports.up = async function up(knex) {
*/
exports.down = async function down(knex) {
if (knex.client.config.client === 'pg') {
await knex.schema.dropTable('events');
await knex.schema.dropTable('event_bus_subscriptions');
await knex.schema.dropTable('event_bus_events');
}
};
plugins/events-backend/src/service/EventsPlugin.test.ts
+8 -8
View File
@@ -150,7 +150,7 @@ describe('eventsPlugin', () => {
'should be possible to publish events as a service, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
@@ -177,7 +177,7 @@ describe('eventsPlugin', () => {
'should be possible to subscribe as a service and receive an event, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
@@ -217,7 +217,7 @@ describe('eventsPlugin', () => {
'should only send an event for each subscriber once, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
@@ -264,7 +264,7 @@ describe('eventsPlugin', () => {
'should not notify subscribers that have already consumed the event, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
@@ -309,7 +309,7 @@ describe('eventsPlugin', () => {
'should return multiple events in order, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
@@ -355,7 +355,7 @@ describe('eventsPlugin', () => {
'should skip publishing if all subscribers have already consumed the event, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
@@ -373,7 +373,7 @@ describe('eventsPlugin', () => {
'should time out when no events are available, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
await helper.subscribe('tester', ['test']).expect(201);
@@ -423,7 +423,7 @@ describe('eventsPlugin', () => {
'should refuse listen without a subscription, %p',
async databaseId => {
const backend = await startTestBackend({
features: [eventsPlugin(), await mockKnexFactory(databaseId)],
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
+12 -11
View File
@@ -26,7 +26,7 @@ import {
import { ForwardedError, NotFoundError } from '@backstage/errors';
import { HumanDuration, durationToMilliseconds } from '@backstage/types';
const WINDOW_SIZE_DEFAULT = 10_000;
const WINDOW_MAX_COUNT_DEFAULT = 10_000;
const WINDOW_MIN_AGE_DEFAULT = { minutes: 10 };
const WINDOW_MAX_AGE_DEFAULT = { days: 1 };
@@ -254,7 +254,7 @@ export class DatabaseEventBusStore implements EventBusStore {
/** Events outside of this age will always be deleted */
maxAge?: HumanDuration;
/** Events outside of this count will be deleted if they are outside the minAge window */
size?: number;
maxCount?: number;
};
}): Promise<DatabaseEventBusStore> {
const db = await options.database.getClient();
@@ -278,12 +278,12 @@ export class DatabaseEventBusStore implements EventBusStore {
db,
options.logger,
listener,
options.window?.size ?? WINDOW_SIZE_DEFAULT,
options.window?.maxCount ?? WINDOW_MAX_COUNT_DEFAULT,
durationToMilliseconds(options.window?.minAge ?? WINDOW_MIN_AGE_DEFAULT),
durationToMilliseconds(options.window?.maxAge ?? WINDOW_MAX_AGE_DEFAULT),
);
options.scheduler.scheduleTask({
await options.scheduler.scheduleTask({
id: 'event-bus-cleanup',
frequency: { seconds: 10 },
timeout: { minutes: 1 },
@@ -301,7 +301,7 @@ export class DatabaseEventBusStore implements EventBusStore {
readonly #db: Knex;
readonly #logger: LoggerService;
readonly #listener: DatabaseEventBusListener;
readonly #windowSize: number;
readonly #windowMaxCount: number;
readonly #windowMinAge: number;
readonly #windowMaxAge: number;
@@ -309,14 +309,14 @@ export class DatabaseEventBusStore implements EventBusStore {
db: Knex,
logger: LoggerService,
listener: DatabaseEventBusListener,
windowSize: number,
windowMaxCount: number,
windowMinAge: number,
windowMaxAge: number,
) {
this.#db = db;
this.#logger = logger;
this.#listener = listener;
this.#windowSize = windowSize;
this.#windowMaxCount = windowMaxCount;
this.#windowMinAge = windowMinAge;
this.#windowMaxAge = windowMaxAge;
}
@@ -393,9 +393,10 @@ export class DatabaseEventBusStore implements EventBusStore {
id,
updated_at: this.#db.fn.now(),
topics,
read_until: this.#db.raw(
`( SELECT COALESCE(MAX("id"), 0) FROM "${TABLE_EVENTS}" )`,
),
// TODO(Rugvip): Might be that there's a more performant way to do this
read_until: this.#db.raw(`( SELECT COALESCE(MAX("id"), 0) FROM ?? )`, [
TABLE_EVENTS,
]),
})
.onConflict('id')
.merge(['topics', 'updated_at'])
@@ -523,7 +524,7 @@ export class DatabaseEventBusStore implements EventBusStore {
.select('id')
.from(TABLE_EVENTS)
.orderBy('id', 'desc')
.offset(this.#windowSize),
.offset(this.#windowMaxCount),
)
.wrap('ANY(ARRAY(', '))'),
)
plugins/events-backend/src/service/hub/createEventBusRouter.ts
+19 -21
View File
@@ -114,6 +114,23 @@ cause unnecessary storage of events.
*/
async function createEventBusStore(deps: {
logger: LoggerService;
database: DatabaseService;
scheduler: SchedulerService;
lifecycle: LifecycleService;
httpAuth: HttpAuthService;
}): Promise<EventBusStore> {
const db = await deps.database.getClient();
if (db.client.config.client === 'pg') {
deps.logger.info('Database is PostgreSQL, using database store');
return await DatabaseEventBusStore.create(deps);
}
deps.logger.info('Database is not PostgreSQL, using memory store');
return new MemoryEventBusStore();
}
/**
* Creates a new event bus router
* @internal
@@ -126,30 +143,11 @@ export async function createEventBusRouter(options: {
httpAuth: HttpAuthService;
notifyTimeoutMs?: number; // for testing
}): Promise<Handler> {
const {
database,
httpAuth,
scheduler,
lifecycle,
notifyTimeoutMs = DEFAULT_NOTIFY_TIMEOUT_MS,
} = options;
const { httpAuth, notifyTimeoutMs = DEFAULT_NOTIFY_TIMEOUT_MS } = options;
const logger = options.logger.child({ type: 'EventBus' });
const router = Router();
let store: EventBusStore;
const db = await database.getClient();
if (db.client.config.client === 'pg') {
logger.info('Database is PostgreSQL, using database store');
store = await DatabaseEventBusStore.create({
database,
logger,
scheduler,
lifecycle,
});
} else {
logger.info('Database is not PostgreSQL, using memory store');
store = new MemoryEventBusStore();
}
const store = await createEventBusStore(options);
const apiRouter = await createOpenApiRouter();