Ensure audio frame when publishing (#690)

Ensure at least one audio frame was generated before returning
publish().

Author: hiroshihorie

.nanpa/publish-audio-check-frames.kdl

@@ -0,0 +1 @@
+patch type="changed" "Ensure audio frames are being generated when publishing audio"

Sources/LiveKit/Core/Room+Engine.swift

@@ -367,7 +367,6 @@ extension Room {
                                         self.log("[Connect] Retry cycle waiting for \(String(format: "%.2f", delay)) seconds before attempt \(attempt + 1)")
                                         return delay
                                     }) { currentAttempt, totalAttempts in
-
                 // Not reconnecting state anymore
                 guard let currentMode = self._state.isReconnectingWithMode else {
                     self.log("[Connect] Not in reconnect state anymore, exiting retry cycle.")

Sources/LiveKit/Core/Room+EngineDelegate.swift

@@ -103,7 +103,6 @@ extension Room {
 
     func engine(_ engine: Room, didUpdateSpeakers speakers: [Livekit_SpeakerInfo]) {
         let activeSpeakers = _state.mutate { state -> [Participant] in
-
             var activeSpeakers: [Participant] = []
             var seenParticipantSids = [Participant.Sid: Bool]()
             for speaker in speakers {

Sources/LiveKit/Core/Room+SignalClientDelegate.swift

@@ -143,7 +143,6 @@ extension Room: SignalClientDelegate {
         log("speakers: \(speakers)", .trace)
 
         let activeSpeakers = _state.mutate { state -> [Participant] in
-
             var lastSpeakers = state.activeSpeakers.reduce(into: [Sid: Participant]()) { $0[$1.sid] = $1 }
             for speaker in speakers {
                 let participantSid = Participant.Sid(from: speaker.sid)

Sources/LiveKit/Core/Room.swift

@@ -242,7 +242,6 @@ public class Room: NSObject, @unchecked Sendable, ObservableObject, Loggable {
 
         // trigger events when state mutates
         _state.onDidMutate = { [weak self] newState, oldState in
-
             guard let self else { return }
 
             // sid updated

Sources/LiveKit/DataStream/Outgoing/StreamWriterDestination.swift

@@ -18,6 +18,6 @@ import Foundation
 
 protocol StreamWriterDestination: Sendable {
     var isOpen: Bool { get async }
-    func write<T: StreamData>(_ data: T) async throws
+    func write(_ data: some StreamData) async throws
     func close(reason: String?) async throws
 }

Sources/LiveKit/Participant/LocalParticipant.swift

@@ -568,7 +568,6 @@ extension LocalParticipant {
                 let videoLayers = dimensions.videoLayers(for: encodings)
 
                 populatorFunc = { populator in
-
                     self.log("[publish] using layers: \(videoLayers.map { String(describing: $0) }.joined(separator: ", "))")
 
                     var simulcastCodecs: [Livekit_SimulcastCodec] = [
@@ -696,6 +695,12 @@ extension LocalParticipant {
                 }
             }()
 
+            // At this point at least 1 audio frame should be generated to continue
+            if let track = track as? LocalAudioTrack {
+                log("[Publish] Waiting for audio frame...")
+                try await track.startWaitingForFrames()
+            }
+
             if track is LocalVideoTrack {
                 if let firstCodecMime = trackInfo.codecs.first?.mimeType,
                    let firstVideoCodec = VideoCodec.from(mimeType: firstCodecMime)

Sources/LiveKit/Participant/Participant.swift

@@ -122,7 +122,6 @@ public class Participant: NSObject, @unchecked Sendable, ObservableObject, Logga
 
         // trigger events when state mutates
         _state.onDidMutate = { [weak self] newState, oldState in
-
             guard let self, let room = self._room else { return }
 
             if newState.isSpeaking != oldState.isSpeaking {

Sources/LiveKit/Support/AsyncCompleter.swift

@@ -132,12 +132,12 @@ final class AsyncCompleter<T: Sendable>: @unchecked Sendable, Loggable {
     }
 
     public func resume(returning value: T) {
-        log("\(label)")
+        log("\(label)", .trace)
         resume(with: .success(value))
     }
 
     public func resume(throwing error: Error) {
-        log("\(label)")
+        log("\(label)", .error)
         resume(with: .failure(error))
     }
 
@@ -161,7 +161,6 @@ final class AsyncCompleter<T: Sendable>: @unchecked Sendable, Loggable {
         // Create a cancel-aware timed continuation
         return try await withTaskCancellationHandler {
             try await withUnsafeThrowingContinuation { continuation in
-
                 // Create time-out block
                 let timeoutBlock = DispatchWorkItem { [weak self] in
                     guard let self else { return }

Sources/LiveKit/Track/Local/LocalAudioTrack.swift

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+import AVFAudio
 import Combine
 import Foundation
 
@@ -26,7 +27,15 @@ internal import LiveKitWebRTC
 @objc
 public class LocalAudioTrack: Track, LocalTrack, AudioTrack, @unchecked Sendable {
     /// ``AudioCaptureOptions`` used to create this track.
-    let captureOptions: AudioCaptureOptions
+    public let captureOptions: AudioCaptureOptions
+
+    // MARK: - Internal
+
+    struct FrameWatcherState {
+        var frameWatcher: AudioFrameWatcher?
+    }
+
+    let _frameWatcherState = StateSync(FrameWatcherState())
 
     init(name: String,
          source: Track.Source,
@@ -43,6 +52,10 @@ public class LocalAudioTrack: Track, LocalTrack, AudioTrack, @unchecked Sendable
                    reportStatistics: reportStatistics)
     }
 
+    deinit {
+        cleanUpFrameWatcher()
+    }
+
     public static func createTrack(name: String = Track.microphoneName,
                                    options: AudioCaptureOptions? = nil,
                                    reportStatistics: Bool = false) -> LocalAudioTrack
@@ -93,6 +106,10 @@ public class LocalAudioTrack: Track, LocalTrack, AudioTrack, @unchecked Sendable
             throw error
         }
     }
+
+    override func stopCapture() async throws {
+        cleanUpFrameWatcher()
+    }
 }
 
 public extension LocalAudioTrack {
@@ -109,3 +126,49 @@ public extension LocalAudioTrack {
         AudioManager.shared.remove(localAudioRenderer: audioRenderer)
     }
 }
+
+// MARK: - Internal frame waiting
+
+extension LocalAudioTrack {
+    final class AudioFrameWatcher: AudioRenderer, Loggable {
+        private let completer = AsyncCompleter<Void>(label: "Frame watcher", defaultTimeout: 5)
+
+        func wait() async throws {
+            try await completer.wait()
+        }
+
+        func reset() {
+            completer.reset()
+        }
+
+        // MARK: - AudioRenderer
+
+        func render(pcmBuffer _: AVAudioPCMBuffer) {
+            completer.resume(returning: ())
+        }
+    }
+
+    func startWaitingForFrames() async throws {
+        let frameWatcher = _frameWatcherState.mutate {
+            $0.frameWatcher?.reset()
+            let watcher = AudioFrameWatcher()
+            add(audioRenderer: watcher)
+            $0.frameWatcher = watcher
+            return watcher
+        }
+
+        try await frameWatcher.wait()
+        // Detach after wait is complete
+        cleanUpFrameWatcher()
+    }
+
+    func cleanUpFrameWatcher() {
+        _frameWatcherState.mutate {
+            if let watcher = $0.frameWatcher {
+                watcher.reset()
+                remove(audioRenderer: watcher)
+                $0.frameWatcher = nil
+            }
+        }
+    }
+}