diff --git a/apps/helper/Sources/SimBLEHelperKit/LoopbackListener.swift b/apps/helper/Sources/SimBLEHelperKit/LoopbackListener.swift index 4324f31..5d533eb 100644 --- a/apps/helper/Sources/SimBLEHelperKit/LoopbackListener.swift +++ b/apps/helper/Sources/SimBLEHelperKit/LoopbackListener.swift @@ -145,9 +145,9 @@ public final class LoopbackListener: @unchecked Sendable { defer { writeLock.unlock() } try? writeFrame(fd, Wire.encode(event)) } - router.attachEventSink(emit) + let sinkID = router.attachEventSink(emit) defer { - router.detachEventSink() + router.detachEventSink(sinkID) close(fd) } while true { diff --git a/apps/helper/Sources/SimBLEHelperKit/RequestRouter.swift b/apps/helper/Sources/SimBLEHelperKit/RequestRouter.swift index 8115c6e..621af80 100644 --- a/apps/helper/Sources/SimBLEHelperKit/RequestRouter.swift +++ b/apps/helper/Sources/SimBLEHelperKit/RequestRouter.swift @@ -15,19 +15,20 @@ enum BridgeErrorCode { } /// Turns a request into a response by driving the central service or the peripheral -/// service, behind the capability-token gate, and streams both services' events to the +/// service, behind the capability-token gate, and streams both services' events to every /// connected client. One service of each role is shared across connections; the listener -/// attaches a per-connection event sink for the duration of a connection. +/// registers a per-connection event sink for the duration of a connection. public final class RequestRouter: @unchecked Sendable { private let service: CentralService private let peripheralService: PeripheralService private let gate: AuthGate private let sinkLock = NSLock() - private var eventSink: (@Sendable (Event) -> Void)? + private var eventSinks: [UInt64: @Sendable (Event) -> Void] = [:] + private var nextSinkID: UInt64 = 0 /// Build the router over the central service, the peripheral service, and the token - /// gate. The router installs each service's event sink once and forwards every event - /// to the attached connection. + /// gate. The router installs each service's event sink once and fans every event out to + /// the registered connections. public init(service: CentralService, peripheralService: PeripheralService, gate: AuthGate) { self.service = service self.peripheralService = peripheralService @@ -35,26 +36,28 @@ public final class RequestRouter: @unchecked Sendable { let forward: @Sendable (Event) -> Void = { [weak self] event in guard let self else { return } sinkLock.lock() - let sink = eventSink + let sinks = Array(eventSinks.values) sinkLock.unlock() - sink?(event) + for sink in sinks { sink(event) } } service.onEvent(forward) peripheralService.onEvent(forward) } - /// Route both services' events to `sink` for the life of one connection. The - /// listener attaches before serving and detaches after. - public func attachEventSink(_ sink: @escaping @Sendable (Event) -> Void) { + /// Register `sink` to receive both services' events. Returns the id that removes it. + public func attachEventSink(_ sink: @escaping @Sendable (Event) -> Void) -> UInt64 { sinkLock.lock() - eventSink = sink + let id = nextSinkID + nextSinkID += 1 + eventSinks[id] = sink sinkLock.unlock() + return id } - /// Stop routing events to the connection's sink. - public func detachEventSink() { + /// Remove the sink registered under `id`. + public func detachEventSink(_ id: UInt64) { sinkLock.lock() - eventSink = nil + eventSinks[id] = nil sinkLock.unlock() } diff --git a/apps/helper/Tests/SimBLEHelperKitTests/RouterFanOutTests.swift b/apps/helper/Tests/SimBLEHelperKitTests/RouterFanOutTests.swift new file mode 100644 index 0000000..ebc783f --- /dev/null +++ b/apps/helper/Tests/SimBLEHelperKitTests/RouterFanOutTests.swift @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: 2026 Nirapod Labs + +import Foundation +@testable import SimBLEHelperKit +import SimBLEHostCore +import SimBLEProtocol +import XCTest + +/// The router fans one backend event out to every registered sink, and a detached sink +/// stops receiving. Driven through a fake backend with no radio. +final class RouterFanOutTests: XCTestCase { + private let peripheralId = Data([0xDE, 0xAD, 0xBE, 0xEF]) + + private let discovery = CentralBackendEvent.discovered( + peripheralId: Data([0xDE, 0xAD, 0xBE, 0xEF]), localName: "Sensor", + serviceUUIDs: ["180D"], txPower: nil, manufacturerData: nil, rssi: -40 + ) + + private var expectedEvent: Event { + .discovered(peripheralId: peripheralId, + advertisement: Advertisement(localName: "Sensor", serviceUUIDs: ["180D"]), + rssi: -40) + } + + private func router(backend: FakeCentralBackend) -> RequestRouter { + RequestRouter(service: CentralService(backend: backend), + peripheralService: PeripheralService(backend: FakePeripheralBackend()), + gate: AuthGate(session: CapabilityToken())) + } + + /// A recorder of the events a sink receives, safe to read after the emit returns. + private final class Recorder: @unchecked Sendable { + private let lock = NSLock() + private var events: [Event] = [] + + var sink: @Sendable (Event) -> Void { + { event in self.lock.lock(); self.events.append(event); self.lock.unlock() } + } + + var received: [Event] { + lock.lock(); defer { lock.unlock() }; return events + } + } + + func testEventFansOutToEverySink() { + let backend = FakeCentralBackend() + let router = router(backend: backend) + let first = Recorder() + let second = Recorder() + _ = router.attachEventSink(first.sink) + _ = router.attachEventSink(second.sink) + + backend.emit(discovery) + + XCTAssertEqual(first.received, [expectedEvent]) + XCTAssertEqual(second.received, [expectedEvent]) + } + + func testDetachedSinkStopsReceiving() { + let backend = FakeCentralBackend() + let router = router(backend: backend) + let kept = Recorder() + let dropped = Recorder() + _ = router.attachEventSink(kept.sink) + let droppedID = router.attachEventSink(dropped.sink) + + router.detachEventSink(droppedID) + backend.emit(discovery) + + XCTAssertEqual(kept.received, [expectedEvent]) + XCTAssertEqual(dropped.received, []) + } +}