diff --git a/Sources/ForgeCamera/ConnectionSupervisor.swift b/Sources/ForgeCamera/ConnectionSupervisor.swift index b26a841..acda02a 100644 --- a/Sources/ForgeCamera/ConnectionSupervisor.swift +++ b/Sources/ForgeCamera/ConnectionSupervisor.swift @@ -40,7 +40,8 @@ public actor ConnectionSupervisor { public private(set) var health: ConnectionStatus = .disconnected private var stateContinuations: [UUID: AsyncStream.Continuation] = [:] - private var connectionWaiters: [CheckedContinuation] = [] + private var connectionWaiters: [UUID] = [] + private var waiterMap: [UUID: CheckedContinuation] = [:] public init(driver: any CameraDriver, backoff: BackoffPolicy) { self.driver = driver @@ -99,9 +100,7 @@ public actor ConnectionSupervisor { health = state.connection if state.connection == .connected { backoff.reset() - let waiters = connectionWaiters - connectionWaiters.removeAll() - for w in waiters { w.resume() } + resumeAllWaiters() } for c in stateContinuations.values { c.yield(state) @@ -109,22 +108,29 @@ public actor ConnectionSupervisor { } /// Await connected health. Returns false on timeout. + /// Continuation registered synchronously on the actor — no gap for a + /// connected event to slip through unobserved. Timeout resumes the + /// continuation directly (cancellation can't interrupt a checked continuation). public func waitForConnection(timeout: Duration) async -> Bool { if health == .connected { return true } - let waitTask = Task { - await withCheckedContinuation { (c: CheckedContinuation) in - self.connectionWaiters.append(c) + await withCheckedContinuation { (c: CheckedContinuation) in + let id = UUID() + waiterMap[id] = c + connectionWaiters.append(id) + Task { + try? await Task.sleep(for: timeout) + self.timeoutWaiter(id: id) } } - let timeoutTask = Task { - try? await Task.sleep(for: timeout) - waitTask.cancel() - } - _ = await waitTask.value - timeoutTask.cancel() return health == .connected } + private func timeoutWaiter(id: UUID) { + guard let c = waiterMap.removeValue(forKey: id) else { return } + connectionWaiters.removeAll { $0 == id } + c.resume() + } + public func stop() async { connectTask?.cancel() pumpTask?.cancel() @@ -132,8 +138,14 @@ public actor ConnectionSupervisor { pumpTask = nil await driver.disconnect() health = .disconnected - let waiters = connectionWaiters + resumeAllWaiters() + } + + private func resumeAllWaiters() { + let ids = connectionWaiters connectionWaiters.removeAll() - for w in waiters { w.resume() } + for id in ids { + waiterMap.removeValue(forKey: id)?.resume() + } } }