feat(core): make output queueAudioFrame suspendable

Author: ThibaultBee

core/src/androidTest/java/io/github/thibaultbee/streampack/core/pipelines/outputs/StubPipelineOutput.kt

@@ -43,8 +43,8 @@ open class StubAudioAsyncPipelineOutput :
     private val _audioFrameFlow = MutableStateFlow<RawFrame?>(null)
     val audioFrameFlow = _audioFrameFlow.asStateFlow()
 
-    override fun queueAudioFrame(frame: RawFrame) {
-        _audioFrameFlow.value = frame
+    override suspend fun queueAudioFrame(frame: RawFrame) {
+        _audioFrameFlow.emit(frame)
     }
 }
 
@@ -84,8 +84,8 @@ class StubAudioSyncVideoSurfacePipelineOutput(resolution: Size) :
     private val _audioFrameFlow = MutableStateFlow<RawFrame?>(null)
     val audioFrameFlow = _audioFrameFlow.asStateFlow()
 
-    override fun queueAudioFrame(frame: RawFrame) {
-        _audioFrameFlow.value = frame
+    override suspend fun queueAudioFrame(frame: RawFrame) {
+        _audioFrameFlow.emit(frame)
     }
 }
 
@@ -150,8 +150,8 @@ class StubAudioSyncVideoSurfacePipelineOutputInternal(resolution: Size) :
     private val _audioFrameFlow = MutableStateFlow<RawFrame?>(null)
     val audioFrameFlow = _audioFrameFlow.asStateFlow()
 
-    override fun queueAudioFrame(frame: RawFrame) {
-        _audioFrameFlow.value = frame
+    override suspend fun queueAudioFrame(frame: RawFrame) {
+        _audioFrameFlow.emit(frame)
     }
 }
 

core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/IPipelineOutput.kt

@@ -244,7 +244,7 @@ interface IAudioSyncPipelineOutputInternal : IAudioPipelineOutputInternal {
      *
      * @param frame The audio [RawFrame] to queue.
      */
-    fun queueAudioFrame(frame: RawFrame)
+    suspend fun queueAudioFrame(frame: RawFrame)
 }
 
 /**

core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/EncodingPipelineOutput.kt

@@ -52,7 +52,6 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
-import kotlinx.coroutines.cancelChildren
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.consumeEach
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -114,7 +113,13 @@ internal class EncodingPipelineOutput(
     override val surfaceFlow = _surfaceFlow.asStateFlow()
 
     // ENCODERS
+    private var audioInput: IEncoderInternal.ISyncByteBufferInput? = null
+
     private var audioEncoderInternal: IEncoderInternal? = null
+        set(value) {
+            audioInput = value?.input as? IEncoderInternal.ISyncByteBufferInput
+            field = value
+        }
     override val audioEncoder: IEncoder?
         get() = audioEncoderInternal
 
@@ -281,9 +286,8 @@ internal class EncodingPipelineOutput(
         }
     }
 
-    override fun queueAudioFrame(frame: RawFrame) {
-        val encoder = requireNotNull(audioEncoderInternal) { "Audio is not configured" }
-        val input = encoder.input as IEncoderInternal.ISyncByteBufferInput
+    override suspend fun queueAudioFrame(frame: RawFrame) = mutex.withLock {
+        val input = requireNotNull(audioInput) { "Audio input is null" }
         input.queueInputFrame(
             frame
         )