diff --git a/packages/couchdb-changes-stream/src/CouchDBChangesStream.ts b/packages/couchdb-changes-stream/src/CouchDBChangesStream.ts index 8857bae..8f0d2df 100644 --- a/packages/couchdb-changes-stream/src/CouchDBChangesStream.ts +++ b/packages/couchdb-changes-stream/src/CouchDBChangesStream.ts @@ -86,12 +86,18 @@ export class CouchDBChangesStream { let heartbeatTimeout: ReturnType | null = null; let heartbeatPromise: Promise = new Promise(() => {}); + let abortControllerWithHeartbeat: AbortController = new AbortController(); + signal.addEventListener("abort", () => { if (heartbeatTimeout) { clearTimeout(heartbeatTimeout); heartbeatTimeout = null; } + abortControllerWithHeartbeat.abort(signal.reason); }); + if (signal.aborted) { + abortControllerWithHeartbeat.abort(signal.reason); + } const resetHeartbeat = () => { if (!usesHeartbeat) return; @@ -100,6 +106,13 @@ export class CouchDBChangesStream { clearTimeout(heartbeatTimeout); } + if (abortControllerWithHeartbeat.signal.aborted) { + abortControllerWithHeartbeat = new AbortController(); + if (signal.aborted) { + abortControllerWithHeartbeat.abort(signal.reason); + } + } + const heartbeatInterval = typeof this.options.heartbeat === "number" ? this.options.heartbeat @@ -108,11 +121,9 @@ export class CouchDBChangesStream { const adjustedTimeout = heartbeatInterval + timeoutBuffer; heartbeatPromise = new Promise((resolve, reject) => { heartbeatTimeout = setTimeout(() => { - reject( - new Error( - `Heartbeat timeout: no data received from server after ${adjustedTimeout} ms`, - ), - ); + const reason = `Heartbeat timeout: no data received from server after ${adjustedTimeout} ms`; + abortControllerWithHeartbeat.abort(reason); + reject(new Error(reason)); }, adjustedTimeout); if (typeof heartbeatTimeout === "object" && heartbeatTimeout.unref) { @@ -161,7 +172,7 @@ export class CouchDBChangesStream { : {}), }, body, - signal, + signal: abortControllerWithHeartbeat.signal, }), heartbeatPromise, ]);