refactor(core): endpoint: add a way to returns asynchronous throwable that happens on stream

Author: ThibaultBee

core/src/androidTest/java/io/github/thibaultbee/streampack/core/elements/endpoints/DummyEndpoint.kt

@@ -23,6 +23,7 @@ import io.github.thibaultbee.streampack.core.elements.data.FrameWithCloseable
 import io.github.thibaultbee.streampack.core.elements.encoders.CodecConfig
 import io.github.thibaultbee.streampack.core.pipelines.IDispatcherProvider
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.runBlocking
 
@@ -55,6 +56,7 @@ class DummyEndpoint : IEndpointInternal {
 
     override val metrics: Any
         get() = TODO("Not yet implemented")
+    override val throwableFlow: StateFlow<Throwable?> = MutableStateFlow(null).asStateFlow()
 
     override suspend fun open(descriptor: MediaDescriptor) {
         _isOpenFlow.emit(true)

core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/CombineEndpoint.kt

@@ -32,6 +32,7 @@ import kotlinx.coroutines.cancelChildren
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.combineTransform
+import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
 
@@ -91,6 +92,14 @@ open class CombineEndpoint(
             false
         )
 
+    override val throwableFlow: StateFlow<Throwable?> = merge(
+        *endpointInternals.map { it.throwableFlow }.toTypedArray()
+    ).stateIn(
+        coroutineScope,
+        started = SharingStarted.Eagerly,
+        initialValue = null
+    )
+
     /**
      * The union of all endpoints' [IEndpoint.IEndpointInfo].
      */

core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DynamicEndpoint.kt

@@ -32,9 +32,11 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancelChildren
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
@@ -71,6 +73,13 @@ open class DynamicEndpoint(
     private val _isOpenFlow = MutableStateFlow(false)
     override val isOpenFlow: StateFlow<Boolean> = _isOpenFlow.asStateFlow()
 
+    private val throwableFlows = endpointFlow.map { it?.throwableFlow }
+    override val throwableFlow: StateFlow<Throwable?> = throwableFlows.map { it?.value }.stateIn(
+        coroutineScope,
+        started = SharingStarted.Eagerly,
+        initialValue = null
+    )
+
     /**
      * Only available when the endpoint is opened.
      */
@@ -244,7 +253,7 @@ open class DynamicEndpoint(
 
     private fun getRtmpEndpoint(): IEndpointInternal {
         if (rtmpEndpoint == null) {
-            rtmpEndpoint = Endpoints.createRtmpEndpoint(ioDispatcher)
+            rtmpEndpoint = Endpoints.createRtmpEndpoint(defaultDispatcher, ioDispatcher)
         }
         return rtmpEndpoint!!
     }

core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/IEndpoint.kt

@@ -17,8 +17,8 @@ package io.github.thibaultbee.streampack.core.elements.endpoints
 
 import android.content.Context
 import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.MediaDescriptor
-import io.github.thibaultbee.streampack.core.elements.data.FrameWithCloseable
 import io.github.thibaultbee.streampack.core.elements.data.Frame
+import io.github.thibaultbee.streampack.core.elements.data.FrameWithCloseable
 import io.github.thibaultbee.streampack.core.elements.encoders.CodecConfig
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.FileSink
 import io.github.thibaultbee.streampack.core.elements.interfaces.Releasable
@@ -30,6 +30,13 @@ import kotlinx.coroutines.runBlocking
 
 interface IEndpointInternal : IEndpoint, SuspendStreamable,
     SuspendCloseable, Releasable {
+
+    /**
+     * An asynchronous error that occurred during streaming.
+     * Synchronous errors must be thrown directly by the functions.
+     */
+    val throwableFlow: StateFlow<Throwable?>
+
     /**
      * Opens the endpoint.
      * The endpoint must check if the [MediaDescriptor] is supported and if it is not already opened.

core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/MediaMuxerEndpoint.kt

@@ -30,6 +30,7 @@ import io.github.thibaultbee.streampack.core.logger.Logger
 import io.github.thibaultbee.streampack.core.pipelines.IDispatcherProvider
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
@@ -70,6 +71,8 @@ class MediaMuxerEndpoint(
     private val _isOpenFlow = MutableStateFlow(false)
     override val isOpenFlow = _isOpenFlow.asStateFlow()
 
+    override val throwableFlow: StateFlow<Throwable?> = MutableStateFlow(null).asStateFlow()
+
     override suspend fun open(descriptor: MediaDescriptor) = mutex.withLock {
         when (state) {
             State.PENDING_START, State.STARTED -> {

core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/CompositeEndpoint.kt

@@ -27,7 +27,9 @@ import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxer
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.ISinkInternal
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.SinkConfiguration
 import io.github.thibaultbee.streampack.core.pipelines.IDispatcherProvider
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -66,6 +68,8 @@ class CompositeEndpoint(
     override val isOpenFlow: StateFlow<Boolean>
         get() = sink.isOpenFlow
 
+    override val throwableFlow: StateFlow<Throwable?> = MutableStateFlow(null).asStateFlow()
+
     override suspend fun open(descriptor: MediaDescriptor) {
         sink.open(descriptor)
     }

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

@@ -259,10 +259,7 @@ internal class EncodingPipelineOutput(
                 audioEncoderListener.outputChannel.consumeEach { closeableFrame ->
                     try {
                         audioStreamId?.let {
-                            endpointInternal.write(
-                                closeableFrame,
-                                it
-                            )
+                            endpointInternal.write(closeableFrame, it)
                         } ?: Logger.w(TAG, "Audio frame received but audio stream is not set")
                     } catch (t: Throwable) {
                         onInternalError(t)
@@ -276,17 +273,22 @@ internal class EncodingPipelineOutput(
                 videoEncoderListener.outputChannel.consumeEach { closeableFrame ->
                     try {
                         videoStreamId?.let {
-                            endpointInternal.write(
-                                closeableFrame,
-                                it
-                            )
+                            endpointInternal.write(closeableFrame, it)
                         } ?: Logger.w(TAG, "Video frame received but video stream is not set")
                     } catch (t: Throwable) {
                         onInternalError(t)
                     }
                 }
             }
         }
+
+        coroutineScope.launch {
+            endpointInternal.throwableFlow.collect { t ->
+                t?.let {
+                    onInternalError(it)
+                }
+            }
+        }
     }
 
     private val _audioCodecConfigFlow = MutableStateFlow<AudioCodecConfig?>(null)