Sync streams (Protocol 10) (#291)

hiroshihorie · web-flow · commit 146139823983 · 2024-01-08T19:59:55.000+09:00
Reference: livekit/client-sdk-js#881 Protocol is already set to v10.
diff --git a/Sources/LiveKit/Core/Engine+TransportDelegate.swift b/Sources/LiveKit/Core/Engine+TransportDelegate.swift
@@ -54,15 +54,19 @@ extension Engine: TransportDelegate {
     }
 
     func transport(_ transport: Transport, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, streams: [LKRTCMediaStream]) {
-        log("did add track")
+        guard !streams.isEmpty else {
+            log("Received onTrack with no streams!", .warning)
+            return
+        }
+
         if transport.target == .subscriber {
             // execute block when connected
             execute(when: { state, _ in state.connectionState == .connected },
                     // always remove this block when disconnected
                     removeWhen: { state, _ in state.connectionState == .disconnected })
             { [weak self] in
                 guard let self else { return }
-                self.notify { $0.engine(self, didAddTrack: track, rtpReceiver: rtpReceiver, streams: streams) }
+                self.notify { $0.engine(self, didAddTrack: track, rtpReceiver: rtpReceiver, stream: streams.first!) }
             }
         }
     }
diff --git a/Sources/LiveKit/Core/Room+EngineDelegate.swift b/Sources/LiveKit/Core/Room+EngineDelegate.swift
@@ -129,31 +129,21 @@ extension Room: EngineDelegate {
         }
     }
 
-    func engine(_: Engine, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, streams: [LKRTCMediaStream]) {
-        guard !streams.isEmpty else {
-            log("Received onTrack with no streams!", .warning)
-            return
-        }
-
-        let unpacked = streams[0].streamId.unpack()
-        let participantSid = unpacked.sid
-        var trackSid = unpacked.trackId
-        if trackSid == "" {
-            trackSid = track.trackId
-        }
+    func engine(_: Engine, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, stream: LKRTCMediaStream) {
+        let parts = stream.streamId.unpack()
 
         let participant = _state.read {
-            $0.remoteParticipants.values.first { $0.sid == participantSid }
+            $0.remoteParticipants.values.first { $0.sid == parts.participantSid }
         }
 
         guard let participant else {
-            log("RemoteParticipant not found for sid: \(participantSid), remoteParticipants: \(remoteParticipants)", .warning)
+            log("RemoteParticipant not found for sid: \(parts.participantSid), remoteParticipants: \(remoteParticipants)", .warning)
             return
         }
 
         let task = Task.retrying(retryDelay: 0.2) { _, _ in
             // TODO: Only retry for TrackError.state = error
-            try await participant.addSubscribedMediaTrack(rtcTrack: track, rtpReceiver: rtpReceiver, sid: trackSid)
+            try await participant.addSubscribedMediaTrack(rtcTrack: track, rtpReceiver: rtpReceiver, sid: track.trackId)
         }
 
         Task {
diff --git a/Sources/LiveKit/Core/Transport.swift b/Sources/LiveKit/Core/Transport.swift
@@ -218,15 +218,15 @@ extension Transport: LKRTCPeerConnectionDelegate {
 
     func peerConnection(_: LKRTCPeerConnection,
                         didAdd rtpReceiver: LKRTCRtpReceiver,
-                        streams mediaStreams: [LKRTCMediaStream])
+                        streams: [LKRTCMediaStream])
     {
         guard let track = rtpReceiver.track else {
             log("Track is empty for \(target)", .warning)
             return
         }
 
-        log("didAddTrack type: \(type(of: track)), id: \(track.trackId)")
-        notify { $0.transport(self, didAddTrack: track, rtpReceiver: rtpReceiver, streams: mediaStreams) }
+        log("type: \(type(of: track)), track.id: \(track.trackId), streams: \(streams.map { "Stream(hash: \($0.hash), id: \($0.streamId), videoTracks: \($0.videoTracks.count), audioTracks: \($0.audioTracks.count))" })")
+        notify { $0.transport(self, didAddTrack: track, rtpReceiver: rtpReceiver, streams: streams) }
     }
 
     func peerConnection(_: LKRTCPeerConnection,
diff --git a/Sources/LiveKit/Extensions/Primitives.swift b/Sources/LiveKit/Extensions/Primitives.swift
@@ -17,7 +17,7 @@
 import Foundation
 
 extension String {
-    func unpack() -> (sid: Sid, trackId: String) {
+    func unpack() -> (participantSid: Sid, trackId: String) {
         let parts = split(separator: "|")
         if parts.count == 2 {
             return (String(parts[0]), String(parts[1]))
diff --git a/Sources/LiveKit/Participant/LocalParticipant.swift b/Sources/LiveKit/Participant/LocalParticipant.swift
@@ -139,6 +139,13 @@ public class LocalParticipant: Participant {
                     ]
                 }
 
+                if let mediaPublishOptions = publishOptions as? MediaPublishOptions,
+                   let streamName = mediaPublishOptions.streamName
+                {
+                    // Set stream name if specified in options
+                    populator.stream = streamName
+                }
+
                 return transInit
             }
 
diff --git a/Sources/LiveKit/Protocols/EngineDelegate.swift b/Sources/LiveKit/Protocols/EngineDelegate.swift
@@ -21,7 +21,7 @@ import Foundation
 protocol EngineDelegate: AnyObject {
     func engine(_ engine: Engine, didMutateState state: Engine.State, oldState: Engine.State)
     func engine(_ engine: Engine, didUpdateSpeakers speakers: [Livekit_SpeakerInfo])
-    func engine(_ engine: Engine, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, streams: [LKRTCMediaStream])
+    func engine(_ engine: Engine, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, stream: LKRTCMediaStream)
     func engine(_ engine: Engine, didRemoveTrack track: LKRTCMediaStreamTrack)
     func engine(_ engine: Engine, didReceiveUserPacket packet: Livekit_UserPacket)
 }
diff --git a/Sources/LiveKit/Types/Options/AudioPublishOptions.swift b/Sources/LiveKit/Types/Options/AudioPublishOptions.swift
@@ -17,7 +17,7 @@
 import Foundation
 
 @objc
-public class AudioPublishOptions: NSObject, PublishOptions {
+public class AudioPublishOptions: NSObject, MediaPublishOptions {
     @objc
     public let name: String?
 
@@ -28,13 +28,18 @@ public class AudioPublishOptions: NSObject, PublishOptions {
     @objc
     public let dtx: Bool
 
+    @objc
+    public let streamName: String?
+
     public init(name: String? = nil,
                 encoding: AudioEncoding? = nil,
-                dtx: Bool = true)
+                dtx: Bool = true,
+                streamName: String? = nil)
     {
         self.name = name
         self.encoding = encoding
         self.dtx = dtx
+        self.streamName = streamName
     }
 
     // MARK: - Equal
@@ -43,14 +48,16 @@ public class AudioPublishOptions: NSObject, PublishOptions {
         guard let other = object as? Self else { return false }
         return name == other.name &&
             encoding == other.encoding &&
-            dtx == other.dtx
+            dtx == other.dtx &&
+            streamName == other.streamName
     }
 
     override public var hash: Int {
         var hasher = Hasher()
         hasher.combine(name)
         hasher.combine(encoding)
         hasher.combine(dtx)
+        hasher.combine(streamName)
         return hasher.finalize()
     }
 }
diff --git a/Sources/LiveKit/Types/Options/PublishOptions.swift b/Sources/LiveKit/Types/Options/PublishOptions.swift
@@ -20,3 +20,11 @@ import Foundation
 public protocol PublishOptions {
     var name: String? { get }
 }
+
+@objc
+public protocol MediaPublishOptions: PublishOptions {
+    /// Set stream name for the track. Audio and video tracks with the same stream name
+    /// will be placed in the same `MediaStream` and offer better synchronization.
+    /// By default, camera and microphone will be placed in a stream; as would screen_share and screen_share_audio
+    var streamName: String? { get }
+}
diff --git a/Sources/LiveKit/Types/Options/VideoPublishOptions.swift b/Sources/LiveKit/Types/Options/VideoPublishOptions.swift
@@ -17,7 +17,7 @@
 import Foundation
 
 @objc
-public class VideoPublishOptions: NSObject, PublishOptions {
+public class VideoPublishOptions: NSObject, MediaPublishOptions {
     @objc
     public let name: String?
 
@@ -45,14 +45,18 @@ public class VideoPublishOptions: NSObject, PublishOptions {
     @objc
     public let preferredBackupCodec: VideoCodec?
 
+    @objc
+    public let streamName: String?
+
     public init(name: String? = nil,
                 encoding: VideoEncoding? = nil,
                 screenShareEncoding: VideoEncoding? = nil,
                 simulcast: Bool = true,
                 simulcastLayers: [VideoParameters] = [],
                 screenShareSimulcastLayers: [VideoParameters] = [],
                 preferredCodec: VideoCodec? = nil,
-                preferredBackupCodec: VideoCodec? = nil)
+                preferredBackupCodec: VideoCodec? = nil,
+                streamName: String? = nil)
     {
         self.name = name
         self.encoding = encoding
@@ -62,6 +66,7 @@ public class VideoPublishOptions: NSObject, PublishOptions {
         self.screenShareSimulcastLayers = screenShareSimulcastLayers
         self.preferredCodec = preferredCodec
         self.preferredBackupCodec = preferredBackupCodec
+        self.streamName = streamName
     }
 
     // MARK: - Equal
@@ -75,7 +80,8 @@ public class VideoPublishOptions: NSObject, PublishOptions {
             simulcastLayers == other.simulcastLayers &&
             screenShareSimulcastLayers == other.screenShareSimulcastLayers &&
             preferredCodec == other.preferredCodec &&
-            preferredBackupCodec == other.preferredBackupCodec
+            preferredBackupCodec == other.preferredBackupCodec &&
+            streamName == other.streamName
     }
 
     override public var hash: Int {
@@ -88,6 +94,7 @@ public class VideoPublishOptions: NSObject, PublishOptions {
         hasher.combine(screenShareSimulcastLayers)
         hasher.combine(preferredCodec)
         hasher.combine(preferredBackupCodec)
+        hasher.combine(streamName)
         return hasher.finalize()
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -54,15 +54,19 @@ extension Engine: TransportDelegate {`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`func transport(_ transport: Transport, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, streams: [LKRTCMediaStream]) {`
`57`		`- log("did add track")`
	`57`	`+ guard !streams.isEmpty else {`
	`58`	`+ log("Received onTrack with no streams!", .warning)`
	`59`	`+ return`
	`60`	`+ }`
	`61`	`+`
`58`	`62`	`if transport.target == .subscriber {`
`59`	`63`	`// execute block when connected`
`60`	`64`	`execute(when: { state, _ in state.connectionState == .connected },`
`61`	`65`	`// always remove this block when disconnected`
`62`	`66`	`removeWhen: { state, _ in state.connectionState == .disconnected })`
`63`	`67`	`{ [weak self] in`
`64`	`68`	`guard let self else { return }`
`65`		`- self.notify { $0.engine(self, didAddTrack: track, rtpReceiver: rtpReceiver, streams: streams) }`
	`69`	`+ self.notify { $0.engine(self, didAddTrack: track, rtpReceiver: rtpReceiver, stream: streams.first!) }`
`66`	`70`	`}`
`67`	`71`	`}`
`68`	`72`	`}`
Original file line number	Diff line number	Diff line change
`@@ -218,15 +218,15 @@ extension Transport: LKRTCPeerConnectionDelegate {`
`218`	`218`
`219`	`219`	`func peerConnection(_: LKRTCPeerConnection,`
`220`	`220`	`didAdd rtpReceiver: LKRTCRtpReceiver,`
`221`		`- streams mediaStreams: [LKRTCMediaStream])`
	`221`	`+ streams: [LKRTCMediaStream])`
`222`	`222`	`{`
`223`	`223`	`guard let track = rtpReceiver.track else {`
`224`	`224`	`log("Track is empty for \(target)", .warning)`
`225`	`225`	`return`
`226`	`226`	`}`
`227`	`227`
`228`		`- log("didAddTrack type: \(type(of: track)), id: \(track.trackId)")`
`229`		`- notify { $0.transport(self, didAddTrack: track, rtpReceiver: rtpReceiver, streams: mediaStreams) }`
	`228`	`+ log("type: \(type(of: track)), track.id: \(track.trackId), streams: \(streams.map { "Stream(hash: \($0.hash), id: \($0.streamId), videoTracks: \($0.videoTracks.count), audioTracks: \($0.audioTracks.count))" })")`
	`229`	`+ notify { $0.transport(self, didAddTrack: track, rtpReceiver: rtpReceiver, streams: streams) }`
`230`	`230`	`}`
`231`	`231`
`232`	`232`	`func peerConnection(_: LKRTCPeerConnection,`
Original file line number	Diff line number	Diff line change
`@@ -139,6 +139,13 @@ public class LocalParticipant: Participant {`
`139`	`139`	`]`
`140`	`140`	`}`
`141`	`141`
	`142`	`+ if let mediaPublishOptions = publishOptions as? MediaPublishOptions,`
	`143`	`+ let streamName = mediaPublishOptions.streamName`
	`144`	`+ {`
	`145`	`+ // Set stream name if specified in options`
	`146`	`+ populator.stream = streamName`
	`147`	`+ }`
	`148`	`+`
`142`	`149`	`return transInit`
`143`	`150`	`}`
`144`	`151`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ import Foundation`
`21`	`21`	`protocol EngineDelegate: AnyObject {`
`22`	`22`	`func engine(_ engine: Engine, didMutateState state: Engine.State, oldState: Engine.State)`
`23`	`23`	`func engine(_ engine: Engine, didUpdateSpeakers speakers: [Livekit_SpeakerInfo])`
`24`		`- func engine(_ engine: Engine, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, streams: [LKRTCMediaStream])`
	`24`	`+ func engine(_ engine: Engine, didAddTrack track: LKRTCMediaStreamTrack, rtpReceiver: LKRTCRtpReceiver, stream: LKRTCMediaStream)`
`25`	`25`	`func engine(_ engine: Engine, didRemoveTrack track: LKRTCMediaStreamTrack)`
`26`	`26`	`func engine(_ engine: Engine, didReceiveUserPacket packet: Livekit_UserPacket)`
`27`	`27`	`}`