Skip to content

Commit 01f8814

Browse files
MarlonJDMarlonJD
committed
feat: Enhance Room and OpenSSLDTLSApplicationDataTransport with improved media startup and error handling
1 parent d22b794 commit 01f8814

5 files changed

Lines changed: 262 additions & 20 deletions

File tree

‎Sources/LiveKitNative/Core/Room.swift‎

Lines changed: 21 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,12 @@ struct RoomMediaStartupConfiguration: Sendable {
239239
maxDataChannelFragmentPayloadSize: maxDataChannelFragmentPayloadSize,
240240
dataChannelTransportMode: dataChannelTransportMode
241241
),
242+
inboundSTUNResponder: { credentials in
243+
LocalICEUDPSocketInboundSTUNResponder.start(
244+
candidateStore: candidateStore,
245+
localCredentials: credentials
246+
)
247+
},
242248
consentFreshnessPolicy: consentFreshnessPolicy,
243249
consentFreshnessRetryPolicy: consentFreshnessRetryPolicy
244250
)
@@ -793,9 +799,10 @@ public final class Room: @unchecked Sendable {
793799
startSignalLoop()
794800
LiveKitNativeLogging.log(.info, "Room connected.")
795801
} catch {
796-
stopSignalLoop()
802+
let signalLoopTask = stopSignalLoop()
797803
stopDataChannelReceiveLoop()
798804
await signalConnection.close()
805+
await signalLoopTask?.value
799806
clearConnectionContext()
800807
resetPeerConnectionNegotiationState(restartICE: true)
801808
clearLocalParticipantCommandHandler()
@@ -814,11 +821,12 @@ public final class Room: @unchecked Sendable {
814821
public func disconnect() async {
815822
LiveKitNativeLogging.log(.info, "Disconnecting room.")
816823

817-
stopSignalLoop()
824+
let signalLoopTask = stopSignalLoop()
818825
stopDataChannelReceiveLoop()
819826
await sendLeaveIfConnected()
820827
await transition(to: .disconnecting)
821828
await signalConnection.close()
829+
await signalLoopTask?.value
822830
await requestTracker.clear()
823831
clearConnectionContext()
824832
resetPeerConnectionNegotiationState(restartICE: true)
@@ -1157,13 +1165,15 @@ public final class Room: @unchecked Sendable {
11571165
replaceSignalLoopTask(with: task)
11581166
}
11591167

1160-
private func stopSignalLoop() {
1168+
@discardableResult
1169+
private func stopSignalLoop() -> Task<Void, Never>? {
11611170
let task = signalLoopLock.withLock {
11621171
let task = signalLoopTask
11631172
signalLoopTask = nil
11641173
return task
11651174
}
11661175
task?.cancel()
1176+
return task
11671177
}
11681178

11691179
private func replaceSignalLoopTask(with task: Task<Void, Never>) {
@@ -2445,18 +2455,16 @@ public final class Room: @unchecked Sendable {
24452455
}
24462456

24472457
let localCandidates = subscriberLocalICECandidates()
2448-
guard !localCandidates.isEmpty else {
2449-
storeSubscriberMediaStartupError(
2450-
PeerConnectionNegotiationError.missingSelectedICECandidatePair
2458+
guard subscriberPeerConnection.remoteDescriptionContainsRTPMedia else {
2459+
startSubscriberInboundSTUNResponderIfNeeded(
2460+
configuration: subscriberMediaStartupConfiguration
24512461
)
24522462
return
24532463
}
24542464

2455-
guard subscriberMediaStartupConfiguration.mediaDataBinder != nil ||
2456-
subscriberPeerConnection.remoteDescriptionContainsRTPMedia
2457-
else {
2458-
startSubscriberInboundSTUNResponderIfNeeded(
2459-
configuration: subscriberMediaStartupConfiguration
2465+
guard !localCandidates.isEmpty else {
2466+
storeSubscriberMediaStartupError(
2467+
PeerConnectionNegotiationError.missingSelectedICECandidatePair
24602468
)
24612469
return
24622470
}
@@ -2840,9 +2848,8 @@ public final class Room: @unchecked Sendable {
28402848
policy: policy,
28412849
now: Date().timeIntervalSince1970
28422850
)
2843-
do {
2844-
try await Task.sleep(nanoseconds: sleepNanoseconds)
2845-
} catch {
2851+
_ = try? await Task.sleep(nanoseconds: sleepNanoseconds)
2852+
if Task.isCancelled {
28462853
return
28472854
}
28482855
}

‎Sources/LiveKitNativeWebRTC/OpenSSLDTLSSRTPHandshaker.swift‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -153,8 +153,10 @@ package actor OpenSSLDTLSApplicationDataTransport {
153153

154154
var completed = false
155155
for _ in 0..<max(1, receiveAttemptLimit) {
156+
try ensureOpen()
156157
let status = try session.handshakeStep(completed: &completed)
157158
try await flushOutbound()
159+
try ensureOpen()
158160
if completed {
159161
return try session.handshakeResult(
160162
role: role,
@@ -166,6 +168,7 @@ package actor OpenSSLDTLSApplicationDataTransport {
166168
}
167169

168170
let inbound = try await transport.receive()
171+
try ensureOpen()
169172
try session.provide(inbound)
170173
}
171174

@@ -182,12 +185,14 @@ package actor OpenSSLDTLSApplicationDataTransport {
182185
try ensureOpen()
183186

184187
while true {
188+
try ensureOpen()
185189
if let applicationData = try session.readApplicationData(maxByteCount: maxByteCount) {
186190
try await flushOutbound()
187191
return applicationData
188192
}
189193

190194
let inbound = try await transport.receive()
195+
try ensureOpen()
191196
try session.provide(inbound)
192197
try await flushOutbound()
193198
}
@@ -217,7 +222,7 @@ package actor OpenSSLDTLSApplicationDataTransport {
217222
}
218223

219224
private final class OpenSSLDTLSSession {
220-
private let raw: OpaquePointer
225+
private var raw: OpaquePointer?
221226

222227
init(
223228
identity: OpenSSLDTLSIdentityStorage,
@@ -236,10 +241,16 @@ private final class OpenSSLDTLSSession {
236241
}
237242

238243
func close() {
244+
guard let raw else {
245+
return
246+
}
247+
248+
self.raw = nil
239249
lkn_dtls_session_free(raw)
240250
}
241251

242252
func provide(_ datagram: Data) throws {
253+
let raw = try requireRaw()
243254
let status = datagram.withUnsafeBytes { bytes in
244255
lkn_dtls_session_provide_datagram(
245256
raw,
@@ -253,6 +264,7 @@ private final class OpenSSLDTLSSession {
253264
}
254265

255266
func handshakeStep(completed: inout Bool) throws -> Int32 {
267+
let raw = try requireRaw()
256268
var isComplete: Int32 = 0
257269
let status = lkn_dtls_session_do_handshake(raw, &isComplete)
258270
completed = isComplete != 0
@@ -263,6 +275,7 @@ private final class OpenSSLDTLSSession {
263275
}
264276

265277
func writeApplicationData(_ data: Data) throws {
278+
let raw = try requireRaw()
266279
let status = data.withUnsafeBytes { bytes in
267280
lkn_dtls_session_write_application_data(
268281
raw,
@@ -276,6 +289,7 @@ private final class OpenSSLDTLSSession {
276289
}
277290

278291
func readApplicationData(maxByteCount: Int) throws -> Data? {
292+
let raw = try requireRaw()
279293
let capacity = max(1, maxByteCount)
280294
var data = Data(repeating: 0, count: capacity)
281295
var length = 0
@@ -303,6 +317,7 @@ private final class OpenSSLDTLSSession {
303317
}
304318

305319
func outboundDatagrams() throws -> [Data] {
320+
let raw = try requireRaw()
306321
var datagrams: [Data] = []
307322
while true {
308323
let datagram = try OpenSSLDTLSIdentityStorage.copyBuffer { buffer, capacity, outLength in
@@ -319,6 +334,7 @@ private final class OpenSSLDTLSSession {
319334
role: DTLSSRTPRole,
320335
expectedRemoteFingerprint: DTLSSignature
321336
) throws -> DTLSSRTPHandshakeResult {
337+
let raw = try requireRaw()
322338
let profileIdentifier = lkn_dtls_session_selected_srtp_profile(raw)
323339
guard profileIdentifier != 0 else {
324340
throw DTLSSRTPError.missingSelectedSRTPProtectionProfile
@@ -359,6 +375,14 @@ private final class OpenSSLDTLSSession {
359375
remoteFingerprint: remoteFingerprint
360376
)
361377
}
378+
379+
private func requireRaw() throws -> OpaquePointer {
380+
guard let raw else {
381+
throw SecureMediaTransportError.transportClosed
382+
}
383+
384+
return raw
385+
}
362386
}
363387

364388
private extension SRTPProtectionProfile {

‎Tests/LiveKitNativeIntegrationTests/LiveMediaStartupIntegrationTests.swift‎

Lines changed: 33 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,8 @@ final class LiveMediaStartupIntegrationTests: XCTestCase {
99
let roomName = harness.roomName(suffix: "publisher-media")
1010
let subscriberIdentity = "swift-native-pub-media-sub"
1111
let publisherIdentity = "swift-native-pub-media-pub"
12-
let subscriberRoom = liveMediaIntegrationRoom()
13-
let publisherRoom = liveMediaIntegrationRoom()
12+
let subscriberRoom = liveMediaIntegrationRoom(liveKitURL: harness.liveKitURL)
13+
let publisherRoom = liveMediaIntegrationRoom(liveKitURL: harness.liveKitURL)
1414
let subscriberEvents = LiveKitIntegrationEventRecorder()
1515
subscriberRoom.delegate = subscriberEvents
1616

@@ -65,8 +65,8 @@ final class LiveMediaStartupIntegrationTests: XCTestCase {
6565
let roomName = harness.roomName(suffix: "subscriber-media")
6666
let subscriberIdentity = "swift-native-sub-media-sub"
6767
let publisherIdentity = "swift-native-sub-media-pub"
68-
let subscriberRoom = liveMediaIntegrationRoom()
69-
let publisherRoom = liveMediaIntegrationRoom()
68+
let subscriberRoom = liveMediaIntegrationRoom(liveKitURL: harness.liveKitURL)
69+
let publisherRoom = liveMediaIntegrationRoom(liveKitURL: harness.liveKitURL)
7070
let subscriberEvents = LiveKitIntegrationEventRecorder()
7171
subscriberRoom.delegate = subscriberEvents
7272

@@ -148,9 +148,11 @@ private func liveMediaIntegrationRoomOptions() -> RoomOptions {
148148
)
149149
}
150150

151-
private func liveMediaIntegrationRoom() -> Room {
151+
private func liveMediaIntegrationRoom(liveKitURL: URL) -> Room {
152152
let subscriberIdentity = DTLSSRTPIdentity.generated()
153153
let publisherIdentity = DTLSSRTPIdentity.generated()
154+
let hostCandidateAddresses = liveMediaHostCandidateAddresses(for: liveKitURL)
155+
let bindAddress = liveMediaBindAddress(for: liveKitURL)
154156
let subscriberPeerConnection = PeerConnectionCoordinator(
155157
configuration: NativeWebRTCConfiguration(
156158
role: .subscriber,
@@ -170,16 +172,42 @@ private func liveMediaIntegrationRoom() -> Room {
170172
subscriberPeerConnection: subscriberPeerConnection,
171173
publisherPeerConnection: publisherPeerConnection,
172174
subscriberMediaStartupConfiguration: .defaultLiveMediaData(
175+
hostCandidateAddresses: { hostCandidateAddresses },
176+
bindAddress: bindAddress,
173177
localCredentials: {
174178
subscriberPeerConnection.configuration.iceCredentials
175179
},
176180
identity: subscriberIdentity
177181
),
178182
publisherMediaStartupConfiguration: .defaultLiveMediaData(
183+
hostCandidateAddresses: { hostCandidateAddresses },
184+
bindAddress: bindAddress,
179185
localCredentials: {
180186
publisherPeerConnection.configuration.iceCredentials
181187
},
182188
identity: publisherIdentity
183189
)
184190
)
185191
}
192+
193+
private func liveMediaHostCandidateAddresses(for liveKitURL: URL) -> [ICEInterfaceAddress] {
194+
guard let host = liveKitURL.host?.lowercased(),
195+
["localhost", "127.0.0.1"].contains(host)
196+
else {
197+
return ICEHostCandidateGatherer.localInterfaceAddresses()
198+
}
199+
200+
return [
201+
ICEInterfaceAddress(name: "lo0", address: "127.0.0.1", localPreference: 101),
202+
]
203+
}
204+
205+
private func liveMediaBindAddress(for liveKitURL: URL) -> String {
206+
guard let host = liveKitURL.host?.lowercased(),
207+
["localhost", "127.0.0.1"].contains(host)
208+
else {
209+
return "0.0.0.0"
210+
}
211+
212+
return "127.0.0.1"
213+
}

‎Tests/LiveKitNativeTests/DTLSSRTPTests.swift‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -220,6 +220,51 @@ final class DTLSSRTPTests: XCTestCase {
220220
await serverDTLS.close()
221221
}
222222

223+
func testOpenSSLDTLSApplicationDataReceiveThrowsWhenClosedWhileAwaitingDatagram() async throws {
224+
let clientIdentity = DTLSSRTPIdentity.generated()
225+
let serverIdentity = DTLSSRTPIdentity.generated()
226+
let datagrams = PairedDTLSDatagramTransport.makePair()
227+
let clientDTLS = try OpenSSLDTLSApplicationDataTransport(
228+
identity: clientIdentity,
229+
role: .client,
230+
transport: datagrams.client
231+
)
232+
let serverDTLS = try OpenSSLDTLSApplicationDataTransport(
233+
identity: serverIdentity,
234+
role: .server,
235+
transport: datagrams.server
236+
)
237+
238+
async let clientResult = clientDTLS.performHandshake(
239+
role: .client,
240+
expectedRemoteFingerprint: serverIdentity.fingerprint
241+
)
242+
async let serverResult = serverDTLS.performHandshake(
243+
role: .server,
244+
expectedRemoteFingerprint: clientIdentity.fingerprint
245+
)
246+
_ = try await (clientResult, serverResult)
247+
248+
let receiveTask = Task {
249+
try await serverDTLS.receive()
250+
}
251+
try await Task.sleep(nanoseconds: 10_000_000)
252+
253+
await serverDTLS.close()
254+
try await clientDTLS.send(Data([0x42]))
255+
256+
do {
257+
_ = try await withDTLSTestTimeout {
258+
try await receiveTask.value
259+
}
260+
XCTFail("Expected receive to fail after the DTLS application-data transport closes.")
261+
} catch {
262+
XCTAssertEqual(error as? SecureMediaTransportError, .transportClosed)
263+
}
264+
265+
await clientDTLS.close()
266+
}
267+
223268
func testOpenSSLDTLSApplicationDataCarriesStandardsSCTPAssociationPackets() async throws {
224269
let clientIdentity = DTLSSRTPIdentity.generated()
225270
let serverIdentity = DTLSSRTPIdentity.generated()
@@ -799,6 +844,30 @@ private enum PairedDTLSDatagramTransportError: Error {
799844
case missingPeer
800845
}
801846

847+
private func withDTLSTestTimeout<T: Sendable>(
848+
nanoseconds: UInt64 = 1_000_000_000,
849+
operation: @escaping @Sendable () async throws -> T
850+
) async throws -> T {
851+
try await withThrowingTaskGroup(of: T.self) { group in
852+
group.addTask {
853+
try await operation()
854+
}
855+
group.addTask {
856+
try await Task.sleep(nanoseconds: nanoseconds)
857+
throw DTLSTestTimeoutError()
858+
}
859+
860+
guard let value = try await group.next() else {
861+
throw DTLSTestTimeoutError()
862+
}
863+
864+
group.cancelAll()
865+
return value
866+
}
867+
}
868+
869+
private struct DTLSTestTimeoutError: Error {}
870+
802871
private func dtlsCandidate(foundation: String) -> ICECandidate {
803872
ICECandidate(
804873
foundation: foundation,

0 commit comments

Comments
 (0)