diff --git a/Sources/LiveKit/Core/Room.swift b/Sources/LiveKit/Core/Room.swift index e5d412547..6b3f7bb3e 100644 --- a/Sources/LiveKit/Core/Room.swift +++ b/Sources/LiveKit/Core/Room.swift @@ -279,6 +279,7 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable { connectOptions: ConnectOptions? = nil, roomOptions: RoomOptions? = nil) { + _ = LKNet.bootstrap // Ensure manager shared objects are instantiated #if !LK_BENCHMARK DeviceManager.prepare() diff --git a/Sources/LiveKit/LiveKit.swift b/Sources/LiveKit/LiveKit.swift index 0b1755d0e..d9492c29b 100644 --- a/Sources/LiveKit/LiveKit.swift +++ b/Sources/LiveKit/LiveKit.swift @@ -85,6 +85,7 @@ public class LiveKitSDK: NSObject, Loggable { public static func prepare() { // TODO: Add RTC related initializations DeviceManager.prepare() + _ = LKNet.bootstrap } } diff --git a/Sources/LiveKit/Net/LKNetHTTPClient.swift b/Sources/LiveKit/Net/LKNetHTTPClient.swift new file mode 100644 index 000000000..75f309402 --- /dev/null +++ b/Sources/LiveKit/Net/LKNetHTTPClient.swift @@ -0,0 +1,68 @@ +// Copyright 2026 LiveKit (Apache-2.0) +import Foundation +internal import LiveKitUniFFI + +/// Live HTTP transport backing `livekit-net`, over `URLSession`. Byte-faithful: +/// returns raw response bytes; never decodes to string/JSON. +final class LKNetHTTPClientLive: HttpClient, @unchecked Sendable { + private let session: URLSession + + init(session: URLSession = URLSession(configuration: .default)) { + self.session = session + } + + func request(method: HttpMethod, url: String, headers: [Header], body: Data?) async throws -> HttpResponse { + guard let u = URL(string: url) else { throw TransportError.Connection("invalid url: \(url)") } + var req = URLRequest(url: u) + req.httpMethod = method == .get ? "GET" : "POST" + for h in headers { req.addValue(h.value, forHTTPHeaderField: h.name) } + if let body { req.httpBody = body } + + do { + let (data, response): (Data, URLResponse) = try await withCheckedThrowingContinuation { cont in + let task = session.dataTask(with: req) { data, response, error in + if let error { cont.resume(throwing: error) } + else if let data, let response { cont.resume(returning: (data, response)) } + else { cont.resume(throwing: URLError(.badServerResponse)) } + } + task.resume() + } + guard let http = response as? HTTPURLResponse else { + throw TransportError.Other("non-HTTP response") + } + let respHeaders: [Header] = http.allHeaderFields.compactMap { key, value in + guard let name = key as? String else { return nil } + return Header(name: name, value: "\(value)") + } + return HttpResponse(status: UInt16(clamping: http.statusCode), headers: respHeaders, body: data) + } catch let t as TransportError { + throw t + } catch let e as URLError { + switch e.code { + case .timedOut: throw TransportError.Timeout + case .cancelled: throw TransportError.Closed + default: throw TransportError.Connection(e.localizedDescription) + } + } catch { + throw TransportError.Connection("\(error)") + } + } +} + +/// Forwarding HTTP client registered with `livekit-net`. Production forwards to +/// `LKNetHTTPClientLive`; `#if DEBUG` tests swap the delegate to a dummy. +final class LKNetHTTPClient: HttpClient, @unchecked Sendable { + private let delegate: StateSync + + init(_ initial: any HttpClient = LKNetHTTPClientLive()) { + delegate = StateSync(initial) + } + + #if DEBUG + func setDelegate(_ d: any HttpClient) { delegate.mutate { $0 = d } } + #endif + + func request(method: HttpMethod, url: String, headers: [Header], body: Data?) async throws -> HttpResponse { + try await delegate.read { $0 }.request(method: method, url: url, headers: headers, body: body) + } +} diff --git a/Sources/LiveKit/Net/LKNetWSClient.swift b/Sources/LiveKit/Net/LKNetWSClient.swift new file mode 100644 index 000000000..53ee6de0b --- /dev/null +++ b/Sources/LiveKit/Net/LKNetWSClient.swift @@ -0,0 +1,133 @@ +// Copyright 2026 LiveKit (Apache-2.0) +import Foundation +internal import LiveKitUniFFI + +/// Live WebSocket transport backing `livekit-net`, over `URLSessionWebSocketTask`. +final class LKNetWSClientLive: WsClient, @unchecked Sendable { + func connect(url: String, headers: [Header], timeoutMs: UInt64) async throws -> WsConnectResult { + guard let u = URL(string: url) else { throw TransportError.Connection("invalid url: \(url)") } + let conn = try await LKNetWSConnection(url: u, headers: headers, timeoutMs: timeoutMs) + return WsConnectResult(connection: conn) + } +} + +/// Forwarding WS client registered with `livekit-net`. Production forwards to +/// `LKNetWSClientLive`; `#if DEBUG` tests swap the delegate to a dummy. +final class LKNetWSClient: WsClient, @unchecked Sendable { + private let delegate: StateSync + + init(_ initial: any WsClient = LKNetWSClientLive()) { + delegate = StateSync(initial) + } + + #if DEBUG + func setDelegate(_ d: any WsClient) { delegate.mutate { $0 = d } } + #endif + + func connect(url: String, headers: [Header], timeoutMs: UInt64) async throws -> WsConnectResult { + try await delegate.read { $0 }.connect(url: url, headers: headers, timeoutMs: timeoutMs) + } +} + +/// One open WebSocket for `livekit-net`. Binary frames only; string frames are +/// decoded as UTF-8 bytes at the boundary (Swift does not interpret content). +final class LKNetWSConnection: WsConnection, @unchecked Sendable { + private let session: URLSession + private let task: URLSessionWebSocketTask + private let connectDelegate: WSConnectDelegate + + init(url: URL, headers: [Header], timeoutMs: UInt64) async throws { + var req = URLRequest(url: url, cachePolicy: .useProtocolCachePolicy, + timeoutInterval: TimeInterval(timeoutMs) / 1000.0) + for h in headers { req.addValue(h.value, forHTTPHeaderField: h.name) } + + connectDelegate = WSConnectDelegate() + session = URLSession(configuration: .default, delegate: connectDelegate, delegateQueue: nil) + task = session.webSocketTask(with: req) + + do { + try await withCheckedThrowingContinuation { (cont: CheckedContinuation) in + connectDelegate.setConnectContinuation(cont) + task.resume() + } + } catch { + session.invalidateAndCancel() + throw Self.map(error) + } + } + + deinit { + task.cancel(with: .normalClosure, reason: nil) + session.invalidateAndCancel() + } + + func send(frame: Data) async throws { + do { + try await withCheckedThrowingContinuation { (cont: CheckedContinuation) in + task.send(.data(frame)) { error in + if let error { cont.resume(throwing: error) } else { cont.resume() } + } + } + } catch { + throw Self.map(error) + } + } + + func recv() async throws -> Data? { + guard task.closeCode == .invalid else { return nil } + do { + let message: URLSessionWebSocketTask.Message = try await withCheckedThrowingContinuation { cont in + task.receive { cont.resume(with: $0) } + } + return Self.decode(message) + } catch { + if task.closeCode != .invalid { return nil } + throw Self.map(error) + } + } + + func close() async { + task.cancel(with: .normalClosure, reason: nil) + session.finishTasksAndInvalidate() + } + + static func decode(_ message: URLSessionWebSocketTask.Message) -> Data { + switch message { + case let .data(d): return d + case let .string(s): return Data(s.utf8) + @unknown default: return Data() + } + } + + private static func map(_ error: Error) -> TransportError { + if let u = error as? URLError { + switch u.code { + case .timedOut: return .Timeout + case .cancelled: return .Closed + default: return .Connection(u.localizedDescription) + } + } + return .Connection("\(error)") + } +} + +/// Bridges `URLSessionWebSocketDelegate` open/close callbacks to a continuation. +/// Mirrors `Support/Network/WebSocket.swift`'s `Delegate`. +private final class WSConnectDelegate: NSObject, URLSessionWebSocketDelegate { + private let continuation = StateSync?>(nil) + + func setConnectContinuation(_ c: CheckedContinuation) { + continuation.mutate { $0 = c } + } + + func urlSession(_: URLSession, webSocketTask _: URLSessionWebSocketTask, didOpenWithProtocol _: String?) { + continuation.mutate { $0?.resume(); $0 = nil } + } + + func urlSession(_: URLSession, task _: URLSessionTask, didCompleteWithError error: Error?) { + continuation.mutate { + if let error { $0?.resume(throwing: error) } else { $0?.resume() } + $0 = nil + } + } +} diff --git a/Sources/LiveKit/Net/NetBootstrap.swift b/Sources/LiveKit/Net/NetBootstrap.swift new file mode 100644 index 000000000..13dfa3300 --- /dev/null +++ b/Sources/LiveKit/Net/NetBootstrap.swift @@ -0,0 +1,23 @@ +// Copyright 2026 LiveKit (Apache-2.0) +internal import LiveKitUniFFI + +/// One-time registration of the SDK's transports with `livekit-net`. +/// `setHttpClient`/`setWsClient` are first-registration-wins; a Swift `static let` +/// guarantees this runs exactly once, thread-safe, before the first connection. +enum LKNet { + #if DEBUG + // Swappable forwarding wrappers so FFI-demo tests can inject dummies. + static let httpClient = LKNetHTTPClient() + static let wsClient = LKNetWSClient() + #endif + + static let bootstrap: Void = { + #if DEBUG + setHttpClient(c: httpClient) + setWsClient(c: wsClient) + #else + setHttpClient(c: LKNetHTTPClientLive()) + setWsClient(c: LKNetWSClientLive()) + #endif + }() +} diff --git a/Tests/LiveKitCoreTests/Net/LKNetHTTPClientLiveTests.swift b/Tests/LiveKitCoreTests/Net/LKNetHTTPClientLiveTests.swift new file mode 100644 index 000000000..6feb8b009 --- /dev/null +++ b/Tests/LiveKitCoreTests/Net/LKNetHTTPClientLiveTests.swift @@ -0,0 +1,113 @@ +// Copyright 2026 LiveKit (Apache-2.0) +import Foundation +import Testing +@testable import LiveKit +import LiveKitUniFFI + +// Uses a dedicated, file-private URLProtocol (not the shared LiveKitTestSupport +// `MockURLProtocol`) so this suite's process-global mock state cannot be reset by +// another suite that also uses `MockURLProtocol` (e.g. `RegionManagerTests`) when +// Swift Testing runs suites in parallel. `.serialized` keeps this suite's own tests +// from racing each other on that state. +@Suite(.serialized) +struct LKNetHTTPClientLiveTests { + private func mockSession() -> URLSession { + let config = URLSessionConfiguration.ephemeral + config.protocolClasses = [LKNetMockURLProtocol.self] + return URLSession(configuration: config) + } + + @Test func getMapsResponse() async throws { + LKNetMockURLProtocol.reset() + defer { LKNetMockURLProtocol.reset() } + LKNetMockURLProtocol.setRequestHandler { req in + #expect(req.httpMethod == "GET") + #expect(req.value(forHTTPHeaderField: "x-req") == "1") + return .init(statusCode: 201, headers: ["x-test": "yes"], body: Data("hi".utf8)) + } + let client = LKNetHTTPClientLive(session: mockSession()) + let resp = try await client.request(method: .get, url: "https://example.test/status", + headers: [Header(name: "x-req", value: "1")], body: nil) + #expect(resp.status == 201) + #expect(resp.body == Data("hi".utf8)) + #expect(resp.headers.contains { $0.name.lowercased() == "x-test" && $0.value == "yes" }) + } + + @Test func timeoutMapsToTransportTimeout() async throws { + LKNetMockURLProtocol.reset() + defer { LKNetMockURLProtocol.reset() } + LKNetMockURLProtocol.setRequestHandler { _ in throw URLError(.timedOut) } + let client = LKNetHTTPClientLive(session: mockSession()) + await #expect(throws: TransportError.self) { + _ = try await client.request(method: .get, url: "https://example.test/slow", headers: [], body: nil) + } + } + + @Test func forwardingDelegatesToInner() async throws { + LKNetMockURLProtocol.reset() + defer { LKNetMockURLProtocol.reset() } + LKNetMockURLProtocol.setRequestHandler { _ in .init(statusCode: 200, headers: [:], body: Data()) } + let fwd = LKNetHTTPClient(LKNetHTTPClientLive(session: mockSession())) + let resp = try await fwd.request(method: .get, url: "https://example.test/f", headers: [], body: nil) + #expect(resp.status == 200) + } +} + +/// Isolated `URLProtocol` for `LKNetHTTPClientLiveTests` only. Its state is process +/// global (as all `URLProtocol` state must be), but no other suite references this +/// class, so it cannot be reset mid-request by a parallel suite. Attached solely via +/// the ephemeral session's `protocolClasses`, so it never intercepts other traffic. +private final class LKNetMockURLProtocol: URLProtocol { + struct Response: Sendable { + let statusCode: Int + let headers: [String: String] + let body: Data + } + + private struct State { + var requestHandler: (@Sendable (URLRequest) throws -> Response)? + } + + private static let _state = StateSync(State()) + + static func setRequestHandler(_ handler: (@Sendable (URLRequest) throws -> Response)?) { + _state.mutate { $0.requestHandler = handler } + } + + static func reset() { + _state.mutate { $0.requestHandler = nil } + } + + override class func canInit(with _: URLRequest) -> Bool { true } + + override class func canonicalRequest(for request: URLRequest) -> URLRequest { request } + + override func startLoading() { + guard let handler = Self._state.read({ $0.requestHandler }) else { + client?.urlProtocol(self, didFailWithError: URLError(.badServerResponse)) + return + } + guard let url = request.url else { + client?.urlProtocol(self, didFailWithError: URLError(.badURL)) + return + } + do { + let mock = try handler(request) + guard let response = HTTPURLResponse(url: url, + statusCode: mock.statusCode, + httpVersion: "HTTP/1.1", + headerFields: mock.headers) + else { + client?.urlProtocol(self, didFailWithError: URLError(.badServerResponse)) + return + } + client?.urlProtocol(self, didReceive: response, cacheStoragePolicy: .notAllowed) + client?.urlProtocol(self, didLoad: mock.body) + client?.urlProtocolDidFinishLoading(self) + } catch { + client?.urlProtocol(self, didFailWithError: error) + } + } + + override func stopLoading() {} +} diff --git a/Tests/LiveKitCoreTests/Net/LKNetWSConnectionTests.swift b/Tests/LiveKitCoreTests/Net/LKNetWSConnectionTests.swift new file mode 100644 index 000000000..fcabbc2d7 --- /dev/null +++ b/Tests/LiveKitCoreTests/Net/LKNetWSConnectionTests.swift @@ -0,0 +1,15 @@ +// Copyright 2026 LiveKit (Apache-2.0) +import Foundation +import Testing +@testable import LiveKit + +struct LKNetWSConnectionTests { + @Test func decodeDataFrame() { + let d = Data([0x01, 0x02, 0x03]) + #expect(LKNetWSConnection.decode(.data(d)) == d) + } + + @Test func decodeStringFrameIsUTF8() { + #expect(LKNetWSConnection.decode(.string("hi")) == Data("hi".utf8)) + } +} diff --git a/Tests/LiveKitCoreTests/Net/NetBootstrapTests.swift b/Tests/LiveKitCoreTests/Net/NetBootstrapTests.swift new file mode 100644 index 000000000..aeb7f3200 --- /dev/null +++ b/Tests/LiveKitCoreTests/Net/NetBootstrapTests.swift @@ -0,0 +1,12 @@ +// Copyright 2026 LiveKit (Apache-2.0) +import Testing +@testable import LiveKit +import LiveKitUniFFI + +struct NetBootstrapTests { + @Test func bootstrapRegistersTransports() { + _ = LKNet.bootstrap + #expect(hasHttpClient()) + #expect(hasWsClient()) + } +} diff --git a/Tests/LiveKitCoreTests/Net/NetFFIRoundTripTests.swift b/Tests/LiveKitCoreTests/Net/NetFFIRoundTripTests.swift new file mode 100644 index 000000000..aaa836e62 --- /dev/null +++ b/Tests/LiveKitCoreTests/Net/NetFFIRoundTripTests.swift @@ -0,0 +1,47 @@ +// Copyright 2026 LiveKit (Apache-2.0) +import Foundation +import Testing +@testable import LiveKit +import LiveKitUniFFI + +// Serialized: the DEBUG delegate seam is process-global mutable state; Swift Testing +// runs @Tests in parallel, so these must not run concurrently with each other. +@Suite(.serialized) +struct NetFFIRoundTripTests { + private final class DummyHTTP: HttpClient, @unchecked Sendable { + func request(method: HttpMethod, url: String, headers: [Header], body: Data?) async throws -> HttpResponse { + HttpResponse(status: 201, headers: [Header(name: "x-test", value: "1")], body: Data("hello".utf8)) + } + } + private final class DummyConn: WsConnection, @unchecked Sendable { + private let store = StateSync(nil) + func send(frame: Data) async throws { store.mutate { $0 = frame } } + func recv() async throws -> Data? { store.mutate { let f = $0; $0 = nil; return f } } + func close() async {} + } + private final class DummyWS: WsClient, @unchecked Sendable { + func connect(url: String, headers: [Header], timeoutMs: UInt64) async throws -> WsConnectResult { + WsConnectResult(connection: DummyConn()) + } + } + + @Test func httpGetRoundTripsThroughFFI() async throws { + _ = LKNet.bootstrap + LKNet.httpClient.setDelegate(DummyHTTP()) + defer { LKNet.httpClient.setDelegate(LKNetHTTPClientLive()) } + + let resp = try await selfTestHttpGet(url: "http://example.test/x") + #expect(resp.status == 201) + #expect(resp.body == Data("hello".utf8)) + #expect(resp.headers.contains { $0.name == "x-test" && $0.value == "1" }) + } + + @Test func wsEchoRoundTripsThroughFFI() async throws { + _ = LKNet.bootstrap + LKNet.wsClient.setDelegate(DummyWS()) + defer { LKNet.wsClient.setDelegate(LKNetWSClientLive()) } + + let echoed = try await selfTestWsEcho(url: "ws://example.test/x", payload: Data("ping".utf8)) + #expect(echoed == Data("ping".utf8)) + } +}