refactor(*): introduce streamer pipeline to enable multiple output

ThibaultBee · Jan 14, 2025 · 701e8f6 · 701e8f6
1 parent 7920b7b
commit 701e8f6
Show file tree

Hide file tree

Showing 22 changed files with 1,778 additions and 663 deletions.
diff --git a/...t/java/io/github/thibaultbee/streampack/core/streamer/state/AudioOnlyStreamerStateTest.kt b/...t/java/io/github/thibaultbee/streampack/core/streamer/state/AudioOnlyStreamerStateTest.kt
@@ -55,14 +55,14 @@ class AudioOnlyStreamerStateTest(private val descriptor: MediaDescriptor) {
 
     // Single method calls
     @Test
-    fun configureAudioOnlyTest() {
+    fun configureAudioOnlyTest() = runTest {
         streamer.setAudioConfig(
             ConfigurationUtils.dummyValidAudioConfig()
         )
     }
 
     @Test
-    fun configureVideoOnlyTest() {
+    fun configureVideoOnlyTest() = runTest {
         try {
             streamer.setVideoConfig(
                 ConfigurationUtils.dummyValidVideoConfig()
@@ -73,7 +73,7 @@ class AudioOnlyStreamerStateTest(private val descriptor: MediaDescriptor) {
     }
 
     @Test
-    fun configureTest() {
+    fun configureTest() = runTest {
         try {
             streamer.setConfig(
                 ConfigurationUtils.dummyValidAudioConfig(),
@@ -85,7 +85,7 @@ class AudioOnlyStreamerStateTest(private val descriptor: MediaDescriptor) {
     }
 
     @Test
-    fun configureErrorTest() {
+    fun configureErrorTest() = runTest {
         try {
             streamer.setAudioConfig(
                 ConfigurationUtils.dummyInvalidAudioConfig()
@@ -105,7 +105,7 @@ class AudioOnlyStreamerStateTest(private val descriptor: MediaDescriptor) {
     }
 
     @Test
-    fun configureReleaseTest() {
+    fun configureReleaseTest() = runTest {
         streamer.setAudioConfig(
             ConfigurationUtils.dummyValidAudioConfig()
         )
@@ -151,7 +151,7 @@ class AudioOnlyStreamerStateTest(private val descriptor: MediaDescriptor) {
     }
 
     @Test
-    fun multipleConfigureTest() {
+    fun multipleConfigureTest() = runTest {
         (0..10).forEach { _ ->
             streamer.setAudioConfig(
                 ConfigurationUtils.dummyValidAudioConfig()

diff --git a/...ndroidTest/java/io/github/thibaultbee/streampack/core/streamer/state/StreamerStateTest.kt b/...ndroidTest/java/io/github/thibaultbee/streampack/core/streamer/state/StreamerStateTest.kt
@@ -48,29 +48,29 @@ abstract class StreamerStateTest(
 
     // Single method calls
     @Test
-    open fun configureAudioOnlyTest() {
+    open fun configureAudioOnlyTest() = runTest {
         streamer.setAudioConfig(
             ConfigurationUtils.dummyValidAudioConfig()
         )
     }
 
     @Test
-    open fun configureVideoOnlyTest() {
+    open fun configureVideoOnlyTest() = runTest {
         streamer.setVideoConfig(
             ConfigurationUtils.dummyValidVideoConfig()
         )
     }
 
     @Test
-    open fun configureTest() {
+    open fun configureTest() = runTest {
         streamer.setConfig(
             ConfigurationUtils.dummyValidAudioConfig(),
             ConfigurationUtils.dummyValidVideoConfig()
         )
     }
 
     @Test
-    open fun configureErrorTest() {
+    open fun configureErrorTest() = runTest {
         try {
             streamer.setConfig(
                 ConfigurationUtils.dummyInvalidAudioConfig(),
@@ -117,7 +117,7 @@ abstract class StreamerStateTest(
     }
 
     @Test
-    open fun configureReleaseTest() {
+    open fun configureReleaseTest() = runTest {
         streamer.setConfig(
             ConfigurationUtils.dummyValidAudioConfig(),
             ConfigurationUtils.dummyValidVideoConfig()
@@ -156,7 +156,7 @@ abstract class StreamerStateTest(
 
     // Stress test
     @Test
-    open fun multipleConfigureTest() {
+    open fun multipleConfigureTest() = runTest {
         (0..10).forEach { _ ->
             streamer.setConfig(
                 ConfigurationUtils.dummyValidAudioConfig(),

diff --git a/.../java/io/github/thibaultbee/streampack/core/elements/processing/video/SurfaceProcessor.kt b/.../java/io/github/thibaultbee/streampack/core/elements/processing/video/SurfaceProcessor.kt
@@ -178,8 +178,10 @@ class SurfaceProcessor(
 
         surfaceOutputs.forEach {
             try {
-                it.updateTransformMatrix(surfaceOutputMatrix, textureMatrix)
-                renderer.render(surfaceTexture.timestamp, surfaceOutputMatrix, it.surface)
+                if (it.isStreaming()) {
+                    it.updateTransformMatrix(surfaceOutputMatrix, textureMatrix)
+                    renderer.render(surfaceTexture.timestamp, surfaceOutputMatrix, it.surface)
+                }
             } catch (e: Exception) {
                 Logger.e(TAG, "Error while rendering frame", e)
             }

diff --git a/...ub/thibaultbee/streampack/core/elements/processing/video/outputs/AbstractSurfaceOutput.kt b/...ub/thibaultbee/streampack/core/elements/processing/video/outputs/AbstractSurfaceOutput.kt
@@ -24,7 +24,8 @@ import io.github.thibaultbee.streampack.core.elements.processing.video.utils.GLU
 
 open class AbstractSurfaceOutput(
     override val surface: Surface,
-    final override val resolution: Size
+    final override val resolution: Size,
+    override val isStreaming: () -> Boolean
 ) : ISurfaceOutput {
     protected val lock = Any()
     protected var isClosed = false
@@ -51,6 +52,7 @@ interface ISurfaceOutput {
     val surface: Surface
     val cropRect: Rect
     val resolution: Size
+    val isStreaming: () -> Boolean
 
     fun updateTransformMatrix(output: FloatArray, input: FloatArray)
 

diff --git a/.../io/github/thibaultbee/streampack/core/elements/processing/video/outputs/SurfaceOutput.kt b/.../io/github/thibaultbee/streampack/core/elements/processing/video/outputs/SurfaceOutput.kt
@@ -31,9 +31,10 @@ import io.github.thibaultbee.streampack.core.elements.utils.extensions.rotate
 class SurfaceOutput(
     surface: Surface,
     resolution: Size,
+    isStreaming: () -> Boolean,
     private val transformationInfo: TransformationInfo
 ) :
-    AbstractSurfaceOutput(surface, resolution) {
+    AbstractSurfaceOutput(surface, resolution, isStreaming) {
 
     private val infoProvider: ISourceInfoProvider
         get() = transformationInfo.infoProvider

diff --git a/...va/io/github/thibaultbee/streampack/core/elements/utils/extensions/StateFlowExtensions.kt b/...va/io/github/thibaultbee/streampack/core/elements/utils/extensions/StateFlowExtensions.kt
@@ -0,0 +1,14 @@
+package io.github.thibaultbee.streampack.core.elements.utils.extensions
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.runningFold
+
+data class History<T>(val previous: T?, val current: T)
+
+fun <T> StateFlow<T>.runningHistory(): Flow<History<T>> =
+    runningFold(
+        initial = null as (History<T>?),
+        operation = { accumulator, new -> History(accumulator?.current, new) }
+    ).filterNotNull()