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

events-backend: refactor listening API to be promise based + return headers early

Browse Source 
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2024-05-25 14:41:48 +02:00
parent 0393323d17
commit d3a7aa3738
5 changed files with 95 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
+41 -42
View File
@@ -82,8 +82,8 @@ class DatabaseEventBusListener {
readonly #listeners = new Set<{
topics: Set<string>;
onNotify: (topicId: string) => void;
onError: () => void;
resolve: (result: { topic: string }) => void;
reject: (error: Error) => void;
}>();
#connPromise?: Promise<InternalDbConnection>;
@@ -95,43 +95,37 @@ class DatabaseEventBusListener {
this.#logger = logger.child({ type: 'DatabaseEventBusListener' });
}
async listen(
async waitForUpdate(
topics: Set<string>,
onNotify: (topicId: string) => void,
onError: () => void,
): Promise<() => void> {
signal: AbortSignal,
): Promise<{ topic: string }> {
if (this.#connTimeout) {
clearTimeout(this.#connTimeout);
this.#connTimeout = undefined;
}
await this.#ensureConnection();
const listener = { topics, onNotify, onError };
this.#listeners.add(listener);
return new Promise<{ topic: string }>((resolve, reject) => {
const listener = { topics, resolve, reject };
this.#listeners.add(listener);
return () => {
this.#listeners.delete(listener);
// If we don't have any listeners, destroy the connection after a timeout
if (this.#listeners.size === 0) {
this.#connTimeout = setTimeout(() => {
this.#connPromise?.then(conn => {
this.#logger.info('Listener connection timed out, destroying');
this.#connPromise = undefined;
this.#destroyConnection(conn);
});
}, LISTENER_CONNECTION_TIMEOUT_MS);
}
};
signal.addEventListener('abort', () => {
this.#listeners.delete(listener);
this.#maybeTimeoutConnection();
});
});
}
#handleNotify(topic: string) {
this.#logger.info(`Listener received notification for topic '${topic}'`);
this.#logger.debug(`Listener received notification for topic '${topic}'`);
for (const l of this.#listeners) {
if (l.topics.has(topic)) {
l.onNotify(topic);
l.resolve({ topic });
this.#listeners.delete(l);
}
}
this.#maybeTimeoutConnection();
}
// We don't try to reconnect on error, instead we notify all listeners and let
@@ -142,7 +136,23 @@ class DatabaseEventBusListener {
error,
);
for (const l of this.#listeners) {
l.onError();
l.reject(new Error('Listener connection failed'));
}
this.#listeners.clear();
this.#maybeTimeoutConnection();
}
#maybeTimeoutConnection() {
// If we don't have any listeners, destroy the connection after a timeout
if (this.#listeners.size === 0 && !this.#connTimeout) {
this.#connTimeout = setTimeout(() => {
this.#connTimeout = undefined;
this.#connPromise?.then(conn => {
this.#logger.info('Listener connection timed out, destroying');
this.#connPromise = undefined;
this.#destroyConnection(conn);
});
}, LISTENER_CONNECTION_TIMEOUT_MS);
}
}
@@ -437,14 +447,12 @@ export class DatabaseEventBusStore implements EventBusStore {
};
}
async listen(
async setupListener(
subscriptionId: string,
options: {
signal: AbortSignal;
onNotify: (topicId: string) => void;
onError: () => void;
},
): Promise<void> {
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
const result = await this.#db<SubscriptionsRow>(TABLE_SUBSCRIPTIONS)
.select('topics')
.where({ id: subscriptionId })
@@ -456,21 +464,12 @@ export class DatabaseEventBusStore implements EventBusStore {
);
}
if (options.signal.aborted) {
return;
}
options.signal.throwIfAborted();
const topics = new Set(result.topics ?? []);
const cancel = await this.#listener.listen(
topics,
options.onNotify,
options.onError,
);
if (options.signal.aborted) {
cancel();
} else {
options.signal.addEventListener('abort', cancel);
}
return {
waitForUpdate: () => this.#listener.waitForUpdate(topics, options.signal),
};
}
#cleanup = async () => {
plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
+25 -21
View File
@@ -28,19 +28,19 @@ export class MemoryEventBusStore implements EventBusStore {
>();
#listeners = new Set<{
topics: Set<string>;
notify(topicId: string): void;
resolve(result: { topic: string }): void;
}>();
async publish(options: {
params: EventParams;
subscriberIds: string[];
}): Promise<{ id: string } | undefined> {
const topicId = options.params.topic;
const topic = options.params.topic;
const subscriberIds = new Set(options.subscriberIds);
let hasOtherSubscribers = false;
for (const sub of this.#subscribers.values()) {
if (sub.topics.has(topicId) && !subscriberIds.has(sub.id)) {
if (sub.topics.has(topic) && !subscriberIds.has(sub.id)) {
hasOtherSubscribers = true;
break;
}
@@ -53,8 +53,9 @@ export class MemoryEventBusStore implements EventBusStore {
this.#events.push({ ...options.params, subscriberIds, seq: nextSeq });
for (const listener of this.#listeners) {
if (listener.topics.has(topicId)) {
listener.notify(topicId);
if (listener.topics.has(topic)) {
listener.resolve({ topic });
this.#listeners.delete(listener);
}
}
return { id: String(nextSeq) };
@@ -97,27 +98,30 @@ export class MemoryEventBusStore implements EventBusStore {
return { events: events.map(event => ({ ...event, seq: undefined })) };
}
async listen(
async setupListener(
subscriptionId: string,
options: {
signal: AbortSignal;
onNotify(topicId: string): void;
onError(): void;
},
): Promise<void> {
if (options.signal.aborted) {
return;
}
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
return {
waitForUpdate: async () => {
options.signal.throwIfAborted();
const sub = this.#subscribers.get(subscriptionId);
if (!sub) {
throw new Error(`Subscription not found`);
}
const listener = { topics: sub.topics, notify: options.onNotify };
this.#listeners.add(listener);
const sub = this.#subscribers.get(subscriptionId);
if (!sub) {
throw new Error(`Subscription not found`);
}
options.signal.addEventListener('abort', () => {
this.#listeners.delete(listener);
});
return new Promise<{ topic: string }>(resolve => {
const listener = { topics: sub.topics, resolve };
this.#listeners.add(listener);
options.signal.addEventListener('abort', () => {
this.#listeners.delete(listener);
});
});
},
};
}
}
plugins/events-backend/src/service/hub/createEventBusRouter.ts
+21 -35
View File
@@ -103,53 +103,22 @@ export async function createEventBusRouter(options: {
});
const id = req.params.subscriptionId;
// Don't notify until we know the outcome of reading events
let resolveShouldNotify: (shouldNotify: boolean) => void;
const shouldNotifyPromise = new Promise<boolean>(resolve => {
resolveShouldNotify = resolve;
});
const controller = new AbortController();
req.on('end', () => controller.abort());
let timeout: NodeJS.Timeout | undefined = undefined;
let notified = false;
const notify = async (status: number) => {
if (!notified) {
clearTimeout(timeout);
notified = true;
if (await shouldNotifyPromise) {
res.status(status).end();
}
}
};
// By setting up the listener first we make sure we don't miss any events
// that are published while reading. If an event is published we'll receive
// a notification, which depending on the outcome of the read we may ignore
await store.listen(id, {
const listener = await store.setupListener(id, {
signal: controller.signal,
onNotify() {
notify(204);
},
onError() {
notify(500);
},
});
// By timing out requests we make sure they don't stall or that events get stuck.
// For the caller there's no difference between a timeout and a
// notifications, either way they should try reading again.
timeout = setTimeout(() => {
notify(204);
const timeout = setTimeout(() => {
controller.abort();
}, notifyTimeoutMs);
shouldNotifyPromise.then(shouldNotify => {
if (!shouldNotify) {
clearTimeout(timeout);
}
});
try {
const { events } = await store.readSubscription(id);
@@ -162,10 +131,27 @@ export async function createEventBusRouter(options: {
if (events.length > 0) {
res.json({ events });
} else {
resolveShouldNotify!(true);
res.status(202);
res.flushHeaders();
try {
const { topic } = await listener.waitForUpdate();
logger.info(
`Received notification for subscription '${id}' for topic '${topic}'`,
{ subject: credentials.principal.subject },
);
} finally {
// A small extra delay ensures a more even spread of events across
// consumers in case some consumers are faster than others
await new Promise(resolve =>
setTimeout(resolve, 1 + Math.random() * 9),
);
res.end();
}
}
} finally {
resolveShouldNotify!(false);
controller.abort();
clearTimeout(timeout);
}
},
);
plugins/events-backend/src/service/hub/types.ts
+2 -4
View File
@@ -26,12 +26,10 @@ export type EventBusStore = {
readSubscription(id: string): Promise<{ events: EventParams[] }>;
listen(
setupListener(
subscriptionId: string,
options: {
signal: AbortSignal;
onNotify(topicId: string): void;
onError(): void;
},
): Promise<void>;
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }>;
};
plugins/events-node/src/api/DefaultEventsService.ts
+6 -3
View File
@@ -205,10 +205,12 @@ class PluginEventsService implements EventsService {
}
backoffMs = POLL_BACKOFF_START_MS;
// 204 arrives after a blocking response and means there are new events
// available or we timed out, either way should should try to read events
// 202 means there were no immediately available events, but the
// response will block until either new events are available or the
// request times out. In both cases we should should try to read events
// immediately again
if (res.status === 204) {
if (res.status === 202) {
await res.body?.getReader()?.closed;
process.nextTick(poll);
} else if (res.status === 200) {
const data = await res.json();
@@ -316,6 +318,7 @@ export class DefaultEventsService implements EventsService {
options &&
new DefaultApiClient({
discoveryApi: options.discovery,
fetchApi: { fetch }, // use native node fetch
});
const logger = options?.logger ?? this.logger;
return new PluginEventsService(