logos-messaging/js-waku
1
0
Fork 0
mirror of https://github.com/logos-messaging/js-waku.git synced 2026-01-11 02:03:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve QueryOnConnect cleanup and abort signal handling

Browse Source
This commit is contained in:
Levente Kiss 2025-10-31 18:38:23 +13:00
parent 705ca38bf3
commit a743ca41bc
1 changed files with 49 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
packages/sdk/src/query_on_connect/query_on_connect.ts
View File

@ -52,6 +52,12 @@ export class QueryOnConnect<
private lastTimeOffline: number;
private readonly forceQueryThresholdMs: number;
private isRunning: boolean = false;
private abortController?: AbortController;
private boundStoreConnectHandler?: (event: CustomEvent<PeerId>) => void;
private boundHealthHandler?: (event: CustomEvent<HealthStatus>) => void;
public constructor(
public decoders: IDecoder<T>[],
public stopIfTrue: (msg: T) => boolean,
@ -71,11 +77,28 @@ export class QueryOnConnect<
}
public start(): void {
if (this.isRunning) {
log.warn("QueryOnConnect already running");
return;
}
log.info("starting query-on-connect service");
this.isRunning = true;
this.abortController = new AbortController();
this.setupEventListeners();
}
public stop(): void {
if (!this.isRunning) {
return;
}
log.info("stopping query-on-connect service");
this.isRunning = false;
if (this.abortController) {
this.abortController.abort();
this.abortController = undefined;
}
this.unsetEventListeners();
}
@ -120,7 +143,8 @@ export class QueryOnConnect<
for await (const page of this._queryGenerator(this.decoders, {
timeStart,
timeEnd,
peerId
peerId,
abortSignal: this.abortController?.signal
})) {
// Await for decoding
const messages = (await Promise.all(page)).filter(
@ -166,33 +190,41 @@ export class QueryOnConnect<
}
private setupEventListeners(): void {
this.boundStoreConnectHandler = (event: CustomEvent<PeerId>) => {
void this.maybeQuery(event.detail).catch((err) =>
log.error("query-on-connect error", err)
);
};
this.boundHealthHandler = this.updateLastOfflineDate.bind(this);
this.peerManagerEventEmitter.addEventListener(
PeerManagerEventNames.StoreConnect,
(event) =>
void this.maybeQuery(event.detail).catch((err) =>
log.error("query-on-connect error", err)
)
this.boundStoreConnectHandler
);
this.wakuEventEmitter.addEventListener(
WakuEvent.Health,
this.updateLastOfflineDate.bind(this)
this.boundHealthHandler
);
}
private unsetEventListeners(): void {
this.peerManagerEventEmitter.removeEventListener(
PeerManagerEventNames.StoreConnect,
(event) =>
void this.maybeQuery(event.detail).catch((err) =>
log.error("query-on-connect error", err)
)
);
if (this.boundStoreConnectHandler) {
this.peerManagerEventEmitter.removeEventListener(
PeerManagerEventNames.StoreConnect,
this.boundStoreConnectHandler
);
this.boundStoreConnectHandler = undefined;
}
this.wakuEventEmitter.removeEventListener(
WakuEvent.Health,
this.updateLastOfflineDate.bind(this)
);
if (this.boundHealthHandler) {
this.wakuEventEmitter.removeEventListener(
WakuEvent.Health,
this.boundHealthHandler
);
this.boundHealthHandler = undefined;
}
}
private updateLastOfflineDate(event: CustomEvent<HealthStatus>): void {