From f4f5872fac2b3fcd6a295446cfeaf853c373ecdf Mon Sep 17 00:00:00 2001 From: HJD Date: Tue, 17 Mar 2026 21:44:15 +0100 Subject: [PATCH] Improvement: HKSV recording stream AbortSignal support and graceful generator termination. --- src/lib/camera/RecordingManagement.ts | 22 ++++++++++++++++++++-- src/lib/controller/CameraController.ts | 4 +++- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/src/lib/camera/RecordingManagement.ts b/src/lib/camera/RecordingManagement.ts index 5b2709b97..70d188277 100644 --- a/src/lib/camera/RecordingManagement.ts +++ b/src/lib/camera/RecordingManagement.ts @@ -941,6 +941,7 @@ class CameraRecordingStream extends EventEmitter implements DataStreamProtocolHa readonly delegate: CameraRecordingDelegate; readonly hdsRequestId: number; readonly streamId: number; + private abortController?: AbortController; private closed = false; eventHandler?: Record = { @@ -997,11 +998,12 @@ class CameraRecordingStream extends EventEmitter implements DataStreamProtocolHa let lastFragmentWasMarkedLast = false; try { - this.generator = this.delegate.handleRecordingStreamRequest(this.streamId); + this.abortController = new AbortController(); + this.generator = this.delegate.handleRecordingStreamRequest(this.streamId, this.abortController.signal); for await (const packet of this.generator) { if (this.closed) { - console.error(`[HDS ${this.connection.remoteAddress}] Delegate yielded fragment after stream ${this.streamId} was already closed!`); + debug("[HDS %s] Delegate yielded fragment after stream %d was already closed.", this.connection.remoteAddress, this.streamId); break; } @@ -1082,6 +1084,7 @@ class CameraRecordingStream extends EventEmitter implements DataStreamProtocolHa } return; } finally { + this.abortController = undefined; this.generator = undefined; if (this.generatorTimeout) { @@ -1137,8 +1140,16 @@ class CameraRecordingStream extends EventEmitter implements DataStreamProtocolHa } private handleClosed(closure: () => void): void { + if (this.closed) { + return; + } + this.closed = true; + // Signal the delegate's generator to stop immediately. This allows delegates with abortable async operations (e.g. pacing delays) to interrupt them and + // return without yielding to a closed stream. + this.abortController?.abort(); + if (this.closingTimeout) { clearTimeout(this.closingTimeout); this.closingTimeout = undefined; @@ -1148,6 +1159,13 @@ class CameraRecordingStream extends EventEmitter implements DataStreamProtocolHa this.connection.removeListener(DataStreamConnectionEvent.CLOSED, this.closeListener); if (this.generator) { + // Signal the generator to terminate via the async generator protocol. The return is queued and takes effect after the generator's current await completes. + // Combined with the AbortSignal above, this ensures the generator terminates promptly even if the delegate doesn't check the signal. We catch rejections + // to prevent unhandled promise warnings if the generator's cleanup throws. + // @ts-expect-error: AsyncGenerator.return() requires a value parameter, but we're signaling termination — no meaningful RecordingPacket to provide. + void this.generator.return().catch((error: Error) => + debug("[HDS %s] Error while closing recording generator for stream %d: %s", this.connection.remoteAddress, this.streamId, error.stack ?? String(error))); + // when this variable is defined, the generator hasn't returned yet. // we start a timeout to uncover potential programming mistakes where we await forever and can't free resources. this.generatorTimeout = setTimeout(() => { diff --git a/src/lib/controller/CameraController.ts b/src/lib/controller/CameraController.ts index 4ec2a2674..e62661ef5 100644 --- a/src/lib/controller/CameraController.ts +++ b/src/lib/controller/CameraController.ts @@ -279,8 +279,10 @@ export interface CameraRecordingDelegate { * NOTE: Don't rely on the streamId for unique identification. Two {@link DataStreamConnection}s might share the same identifier space. * * @param streamId - The streamId of the currently ongoing stream. + * @param signal - An optional {@link AbortSignal} that is aborted when the recording stream is closed. Delegates can use this to interrupt pending async + * operations (e.g. pacing delays) for immediate cleanup, rather than waiting for {@link closeRecordingStream} to be called. */ - handleRecordingStreamRequest(streamId: number): AsyncGenerator; + handleRecordingStreamRequest(streamId: number, signal?: AbortSignal): AsyncGenerator; /** * This method is called once the HomeKit Controller acknowledges the `endOfStream`.