couchdb-changes-stream: add abort requst by heartbeat

This commit is contained in:
2024-12-01 17:47:53 +02:00
parent 9cc03623d9
commit e6ec37b473
@@ -86,12 +86,18 @@ export class CouchDBChangesStream<T> {
let heartbeatTimeout: ReturnType<typeof setTimeout> | null = null; let heartbeatTimeout: ReturnType<typeof setTimeout> | null = null;
let heartbeatPromise: Promise<never> = new Promise<never>(() => {}); let heartbeatPromise: Promise<never> = new Promise<never>(() => {});
let abortControllerWithHeartbeat: AbortController = new AbortController();
signal.addEventListener("abort", () => { signal.addEventListener("abort", () => {
if (heartbeatTimeout) { if (heartbeatTimeout) {
clearTimeout(heartbeatTimeout); clearTimeout(heartbeatTimeout);
heartbeatTimeout = null; heartbeatTimeout = null;
} }
abortControllerWithHeartbeat.abort(signal.reason);
}); });
if (signal.aborted) {
abortControllerWithHeartbeat.abort(signal.reason);
}
const resetHeartbeat = () => { const resetHeartbeat = () => {
if (!usesHeartbeat) return; if (!usesHeartbeat) return;
@@ -100,6 +106,13 @@ export class CouchDBChangesStream<T> {
clearTimeout(heartbeatTimeout); clearTimeout(heartbeatTimeout);
} }
if (abortControllerWithHeartbeat.signal.aborted) {
abortControllerWithHeartbeat = new AbortController();
if (signal.aborted) {
abortControllerWithHeartbeat.abort(signal.reason);
}
}
const heartbeatInterval = const heartbeatInterval =
typeof this.options.heartbeat === "number" typeof this.options.heartbeat === "number"
? this.options.heartbeat ? this.options.heartbeat
@@ -108,11 +121,9 @@ export class CouchDBChangesStream<T> {
const adjustedTimeout = heartbeatInterval + timeoutBuffer; const adjustedTimeout = heartbeatInterval + timeoutBuffer;
heartbeatPromise = new Promise<never>((resolve, reject) => { heartbeatPromise = new Promise<never>((resolve, reject) => {
heartbeatTimeout = setTimeout(() => { heartbeatTimeout = setTimeout(() => {
reject( const reason = `Heartbeat timeout: no data received from server after ${adjustedTimeout} ms`;
new Error( abortControllerWithHeartbeat.abort(reason);
`Heartbeat timeout: no data received from server after ${adjustedTimeout} ms`, reject(new Error(reason));
),
);
}, adjustedTimeout); }, adjustedTimeout);
if (typeof heartbeatTimeout === "object" && heartbeatTimeout.unref) { if (typeof heartbeatTimeout === "object" && heartbeatTimeout.unref) {
@@ -161,7 +172,7 @@ export class CouchDBChangesStream<T> {
: {}), : {}),
}, },
body, body,
signal, signal: abortControllerWithHeartbeat.signal,
}), }),
heartbeatPromise, heartbeatPromise,
]); ]);