JanusConnection, PeerConnection wrapper and PeerConnectionManager

Author: Drejzer

Sources/PrivMXEndpointSwift/Errors/PrivMXEndpointError.swift

@@ -45,7 +45,7 @@ public enum PrivMXEndpointError : Error{
 	case failedDisconnecting(privmx.InternalError)
 	/// Failed to list Contexts.
 	case failedListingContexts(privmx.InternalError)
-
+	
 	/// Failed to get a list of Users from a Context.
 	case failedGettingContextUsers(privmx.InternalError)
 
@@ -169,7 +169,7 @@ public enum PrivMXEndpointError : Error{
 
 	/// Failed to emit a `CustomEvent`.
 	case failedEmittingCustomEvent(privmx.InternalError)
-
+	
 	/// Failed to subscribe for Custom Events.
 	case failedSubscribingForCustomEvents(privmx.InternalError)
 
@@ -179,7 +179,7 @@ public enum PrivMXEndpointError : Error{
 	/// Failed to set a User Verifier.
 	case failedSettingUserVerifier(privmx.InternalError)
 
- 	/// Failed to instantiate ExtKey.
+	/// Failed to instantiate ExtKey.
 	case failedInstantiatingExtKey(privmx.InternalError)
 	/// Failed to derive ExtKey.
 	case failedDerivingExtKey(privmx.InternalError)
@@ -201,7 +201,7 @@ public enum PrivMXEndpointError : Error{
 	case failedVerifyingCompactSignature(privmx.InternalError)
 	/// Failed to check if ExtKey is private.
 	case failedCheckingIfExtKeyIsPrivate(privmx.InternalError)
-
+	
 	/// Failed to convert a PGP Asn1 key to Base58DER.
 	case failedConvertingKeyToBase58DER(privmx.InternalError)
 
@@ -269,7 +269,11 @@ public enum PrivMXEndpointError : Error{
 	case failedDeletingKvdbEntries(privmx.InternalError)
 
 	case failedSyncingFile(privmx.InternalError)
-  
+	
+#if Streams
+	case failedInitializingPeerConnection(privmx.InternalError)
+#endif
+	
 	/// Gets the Message of the error.
 	///
 	///  - Returns: Message of the error
@@ -388,6 +392,10 @@ public enum PrivMXEndpointError : Error{
 					.failedSyncingFile(let err),
 					.failedGeneratingBIP39(let err):
 				return String(err.message)
+#if Streams
+			case .failedInitializingPeerConnection(let err):
+				return String(err.message)
+#endif
 		}
 	}
 
@@ -510,6 +518,10 @@ public enum PrivMXEndpointError : Error{
 					.failedSyncingFile(let err),
 					.failedGeneratingBIP39(let err):
 				return err.code.value
+#if Streams
+			case .failedInitializingPeerConnection(let err):
+				return err.code.value
+#endif
 		}
 	}
 
@@ -631,6 +643,10 @@ public enum PrivMXEndpointError : Error{
 					.failedSyncingFile(let err),
 					.failedGeneratingBIP39(let err):
 				return String(err.name)
+#if Streams
+			case .failedInitializingPeerConnection(let err):
+				return String(err.name)
+#endif
 		}
 	}
 
@@ -752,6 +768,10 @@ public enum PrivMXEndpointError : Error{
 					.failedSyncingFile(let err),
 					.failedGeneratingBIP39(let err):
 				return String(err.description)
+#if Streams
+			case .failedInitializingPeerConnection(let err):
+				return String(err.description)
+#endif
 		}
 	}
 
@@ -877,7 +897,14 @@ public enum PrivMXEndpointError : Error{
 				} else {
 					return nil
 				}
-				
+#if Streams
+			case .failedInitializingPeerConnection(let err):
+				if let scope = err.scope.value{
+					return String(scope)
+				} else {
+					return nil
+				}
+#endif
 		}
 	}
 }

Sources/PrivMXEndpointSwift/Streams/JanusConnection.swift

@@ -14,7 +14,19 @@ import Foundation
 import WebRTC
 import PrivMXEndpointStreamsLow
 
-class JanusConnection{
+public final class JanusConnection: @unchecked Sendable{
+	var peerConnection : PeerConnection
+	var sessionId: Int64
+	var hasSubscriptions : Bool
 
+	init(
+		peerConnection: PeerConnection,
+		sessionId: Int64,
+		hasSubscriptions: Bool
+	) {
+		self.peerConnection = peerConnection
+		self.sessionId = sessionId
+		self.hasSubscriptions = hasSubscriptions
+	}
 }
-#endif
+#endif // Streams

Sources/PrivMXEndpointSwift/Streams/PeerConnection.swift

@@ -12,16 +12,34 @@
 import PrivMXEndpointSwiftNative
 import PrivMXEndpointStreamsLow
 import WebRTC
+import os.lock
 
-public class PeerConnection{
-	let rtcPeerConnection: RTCPeerConnection
-	let rtcPeerConnectionObserver: PmxPeerConnectionObserver
+public enum ConnectionType:Sendable{
+	case Subscriber
+	case Publisher
+}
+
+public class PeerConnection: @unchecked Sendable{
+	enum State{
+		case idle
+		case working
+	}
+	var rtcPeerConnection: RTCPeerConnection
+	var rtcPeerConnectionObserver: PmxPeerConnectionObserver
+	var audioTracks : [String:AudioTrackInfo] = [:]
+	var videoTracks : [String:VideoTrackInfo] = [:]
+	var trackMutex = OSAllocatedUnfairLock<State>(initialState: State.idle)
+	var keys : PMXKeyStore
 
-	init(rtcPeerConnection: RTCPeerConnection, rtcPeerConnectionObserver: PmxPeerConnectionObserver) {
+	init(rtcPeerConnection: RTCPeerConnection,
+		 rtcPeerConnectionObserver: PmxPeerConnectionObserver,
+		 keys: PMXKeyStore
+	) {
 		self.rtcPeerConnection = rtcPeerConnection
 		self.rtcPeerConnectionObserver = rtcPeerConnectionObserver
+		self.keys = keys
 	}
 
 }
 
-#endif
+#endif // Streams

Sources/PrivMXEndpointSwift/Streams/PeerConnectionManager.swift

@@ -12,9 +12,115 @@
 #if Streams
 import Foundation
 import PrivMXEndpointStreamsLow
+import WebRTC
+import os.lock
 
-final class PeerConnectionManager {
-	private var _createPeerCOnnection : ((String) -> PeerConnection)?
+public final class PeerConnectionManager: Sendable {
+	enum State{
+		case reading,writing
+		case idle
+	}
+	private let mutex = OSAllocatedUnfairLock(initialState: State.idle)
+	private let _createPeerConnection : (@Sendable (String) -> PeerConnection)
+	private let _onTrickle : (@Sendable (Int64,String) throws -> Void)
+	nonisolated(unsafe) private var _connections : [String : [ConnectionType:JanusConnection]] = [:]
+	nonisolated private var connections : [String : [ConnectionType:JanusConnection]]{
+		set(val) {
+			mutex.withLockUnchecked{
+				state in
+				state = .writing
+				_connections = val
+				state = .idle
+			}
+		}
+		get {
+			mutex.withLock{
+				state in
+				defer{
+					state = .idle
+				}
+				state = .reading
+				return _connections
+			}
+			
+		}
+	}
 
+	init(
+		_createPeerConnection: @Sendable @escaping (String) -> PeerConnection,
+		_onTrickle: (@Sendable @escaping (Int64,String) throws -> Void)
+	) {
+		self._createPeerConnection = _createPeerConnection
+		self._onTrickle = _onTrickle
+	}
+	
+	public func initializeConnection(
+		in streamRoomId: String,
+		ofType type: ConnectionType,
+		sessionId: Int64 = -1
+	) throws {
+		if connections.contains(where: {$0 == streamRoomId && $1.keys.contains(type)}){
+			throw PrivMXEndpointError.failedInitializingPeerConnection(privmx.InternalError(
+				name: "Already Initialized",
+				message: "JanusConnection with given parameters already initialized",
+				description: "", code: nil, scope: nil))
+		}
+		if !connections.keys.contains(streamRoomId){
+			connections[streamRoomId] = [:]
+		}
+		
+		let pc = _createPeerConnection(streamRoomId)
+		
+		
+		pc.rtcPeerConnectionObserver.setIceCandidateGeneratedCallback({
+			peerConnection,candidate in
+			
+			let roomConnections = self.connections[streamRoomId] ?? [:]
+			let roomConnection = roomConnections[type]
+			if !candidate.sdp.isEmpty,let sessionId = roomConnection?.sessionId, sessionId > -1{
+				var iceCandidate = candidate.sdp
+				try? self._onTrickle(sessionId,iceCandidate)
+			}
+		})
+		
+		connections[streamRoomId]![type] = JanusConnection(
+			peerConnection: pc,
+			sessionId: sessionId,
+			hasSubscriptions: false)
+	}
+	
+	public func updateSessionForConnection(
+		streamRoomId:String,
+		connectionType: ConnectionType,
+		sessionId: Int64
+	) throws -> Void {
+		if !connections.contains(where: {$0.key == streamRoomId}) || nil == connections[streamRoomId]?[connectionType] {
+			try initializeConnection(
+				in: streamRoomId,
+				ofType: connectionType,
+				sessionId: sessionId)
+		}
+		connections[streamRoomId]?[connectionType]?.sessionId = sessionId
+	}
+	
+	public func hasConnection(
+		streamRoomId: String,
+		connectionType: ConnectionType
+	) throws -> Bool {
+		return nil != connections[streamRoomId]?[connectionType]
+	}
+	
+	public func getConnectionWithSession(
+		streamRoomId:String,
+		connectionType: ConnectionType
+	)  throws -> JanusConnection {
+		if !connections.contains(where: {$0.key == streamRoomId})
+			|| nil == connections[streamRoomId]?[connectionType] {
+			try initializeConnection(
+				in: streamRoomId,
+				ofType: connectionType)
+		}
+		return connections[streamRoomId]![connectionType]!
+	}
 }
-#endif
+#endif // Streams

Sources/PrivMXEndpointSwift/Streams/PmxPeerConnectionObserver.swift

@@ -14,7 +14,7 @@ import Foundation
 import WebRTC
 import Synchronization
 
-public class PmxPeerConnectionObserver:NSObject,RTCPeerConnectionDelegate, @unchecked Sendable{
+public final class PmxPeerConnectionObserver:NSObject,RTCPeerConnectionDelegate, @unchecked Sendable{
 
 
 	private var onConnectionSignalingStateChanged: ((RTCPeerConnection,RTCSignalingState)->Void)?
@@ -26,8 +26,8 @@ public class PmxPeerConnectionObserver:NSObject,RTCPeerConnectionDelegate, @unch
 	private var onIceCandidateErrorEvent:((RTCPeerConnection,RTCIceCandidateErrorEvent)->Void)?
 	private var onIceConnectionStateChanged:((RTCPeerConnection,RTCIceConnectionState)->Void)?
 	private var onIceGatheringStateChanged:((RTCPeerConnection,RTCIceGatheringState)->Void)?
-	private var onIceCandindateGenerated:((RTCPeerConnection,RTCIceCandidate)->Void)?
-	private var onIceCandindatesRemoved:((RTCPeerConnection,[RTCIceCandidate])->Void)?
+	private var onIceCandidateGenerated:((RTCPeerConnection,RTCIceCandidate)->Void)?
+	private var onIceCandidatesRemoved:((RTCPeerConnection,[RTCIceCandidate])->Void)?
 
 	private var onDataChannelOpened:((RTCPeerConnection,RTCDataChannel)->Void)?
 	private var onStartedReceiving:((RTCPeerConnection,RTCRtpTransceiver)->Void)?
@@ -81,15 +81,15 @@ public class PmxPeerConnectionObserver:NSObject,RTCPeerConnectionDelegate, @unch
 	){
 		onIceGatheringStateChanged = cb
 	}
-	public func setIceCandindateGeneratedCallback(
+	public func setIceCandidateGeneratedCallback(
 		_ cb :(@Sendable (RTCPeerConnection,RTCIceCandidate)->Void)?
 	){
-		onIceCandindateGenerated = cb
+		onIceCandidateGenerated = cb
 	}
-	public func setIceCandindatesRemovedCallback(
+	public func setIceCandidatesRemovedCallback(
 		_ cb :(@Sendable (RTCPeerConnection,[RTCIceCandidate])->Void)?
 	){
-		onIceCandindatesRemoved = cb
+		onIceCandidatesRemoved = cb
 	}
 
 	public func setDataChannelOpenedCallback(
@@ -171,14 +171,14 @@ public class PmxPeerConnectionObserver:NSObject,RTCPeerConnectionDelegate, @unch
 		_ peerConnection: RTCPeerConnection,
 		didGenerate candidate: RTCIceCandidate
 	) {
-		onIceCandindateGenerated?(peerConnection,candidate)
+		onIceCandidateGenerated?(peerConnection,candidate)
 	}
 
 	public func peerConnection(
 		_ peerConnection: RTCPeerConnection,
 		didRemove candidates: [RTCIceCandidate]
 	) {
-		onIceCandindatesRemoved?(peerConnection,candidates)
+		onIceCandidatesRemoved?(peerConnection,candidates)
 	}
 
 	public func peerConnection(

Sources/PrivMXEndpointSwift/Streams/TrackInfo.swift

@@ -9,6 +9,7 @@
 // limitations under the License.
 //
 
+#if Streams
 import Foundation
 import WebRTC
 import PrivMXEndpointStreamsLow
@@ -24,3 +25,4 @@ struct VideoTrackInfo{
 	var sender: RTCRtpSender
 	var frameCryptor: PMXFrameCryptorTransformer
 }
+#endif // Streams