feat(core): make startStream and stopStream suspendable in case t…

…here is a network issue.
ThibaultBee · Dec 3, 2023 · 3b9dde0 · 3b9dde0
1 parent 795698e
commit 3b9dde0
Show file tree

Hide file tree

Showing 28 changed files with 152 additions and 87 deletions.
diff --git a/...est/java/io/github/thibaultbee/streampack/streamer/testcases/AudioOnlyStreamerTestCase.kt b/...est/java/io/github/thibaultbee/streampack/streamer/testcases/AudioOnlyStreamerTestCase.kt
@@ -41,8 +41,8 @@ abstract class AudioOnlyStreamerTestCase :
             )
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
             streamer.release()
         } catch (e: Exception) {
             Log.e(TAG, "defaultUsageTest: exception: ", e)
@@ -58,8 +58,8 @@ abstract class AudioOnlyStreamerTestCase :
             )
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
             streamer.release()
         } catch (e: Exception) {
             Log.e(TAG, "defaultUsageTest2: exception: ", e)
@@ -149,7 +149,9 @@ abstract class AudioOnlyStreamerTestCase :
             streamer.configure(
                 AndroidUtils.fakeValidAudioConfig()
             )
-            streamer.stopStream()
+            runBlocking {
+                streamer.stopStream()
+            }
         } catch (e: Exception) {
             Log.e(TAG, "configureStopStreamTest: exception: ", e)
             fail("Must be possible to configure/stopStream but catches exception: $e")
@@ -180,8 +182,8 @@ abstract class AudioOnlyStreamerTestCase :
             )
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
         } catch (e: Exception) {
             Log.e(TAG, "startStreamStopStreamTest: exception: ", e)
             fail("Must be possible to startStream/stopStream but catches exception: $e")
@@ -197,8 +199,8 @@ abstract class AudioOnlyStreamerTestCase :
             (0..10).forEach { _ ->
                 runBlocking {
                     streamer.startStream()
+                    streamer.stopStream()
                 }
-                streamer.stopStream()
             }
         } catch (e: Exception) {
             Log.e(TAG, "multipleStartStreamStopStreamTest: exception: ", e)

diff --git a/...idTest/java/io/github/thibaultbee/streampack/streamer/testcases/CameraStreamerTestCase.kt b/...idTest/java/io/github/thibaultbee/streampack/streamer/testcases/CameraStreamerTestCase.kt
@@ -58,8 +58,8 @@ abstract class CameraStreamerTestCase :
             streamer.startPreview(surface)
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
             streamer.stopPreview()
             streamer.release()
         } catch (e: Exception) {
@@ -80,8 +80,8 @@ abstract class CameraStreamerTestCase :
             streamer.startPreview(surface)
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
             streamer.release()
         } catch (e: Exception) {
             Log.e(TAG, "defaultUsageTest2: exception: ", e)
@@ -161,7 +161,9 @@ abstract class CameraStreamerTestCase :
                 AndroidUtils.fakeValidVideoConfig()
             )
             streamer.startPreview(surface)
-            streamer.stopStream()
+            runBlocking {
+                streamer.stopStream()
+            }
         } catch (e: Exception) {
             Log.e(TAG, "startPreviewStopStreamTest: exception: ", e)
             fail("Must be possible to startPreview/stopStream but catches exception: $e")
@@ -216,8 +218,8 @@ abstract class CameraStreamerTestCase :
             streamer.startPreview(surface)
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
         } catch (e: Exception) {
             Log.e(TAG, "startStreamStopStreamTest: exception: ", e)
             fail("Must be possible to startStream/stopStream but catches exception: $e")
@@ -252,8 +254,8 @@ abstract class CameraStreamerTestCase :
             (0..10).forEach { _ ->
                 runBlocking {
                     streamer.startStream()
+                    streamer.stopStream()
                 }
-                streamer.stopStream()
             }
         } catch (e: Exception) {
             Log.e(TAG, "multipleStartStreamStopStreamTest: exception: ", e)

diff --git a/.../androidTest/java/io/github/thibaultbee/streampack/streamer/testcases/StreamerTestCase.kt b/.../androidTest/java/io/github/thibaultbee/streampack/streamer/testcases/StreamerTestCase.kt
@@ -44,8 +44,8 @@ abstract class StreamerTestCase {
             )
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
             streamer.release()
         } catch (e: Exception) {
             Log.e(TAG, "defaultUsageTest: exception: ", e)
@@ -64,8 +64,8 @@ abstract class StreamerTestCase {
             )
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
             streamer.release()
         } catch (e: Exception) {
             Log.e(TAG, "defaultUsageTest2: exception: ", e)
@@ -137,7 +137,9 @@ abstract class StreamerTestCase {
     @Test
     fun stopStreamTest() {
         try {
-            streamer.stopStream()
+            runBlocking {
+                streamer.stopStream()
+            }
         } catch (e: Exception) {
             Log.e(TAG, "stopStreamTest: exception: ", e)
             fail("Must be possible to only stopStream without exception: $e")
@@ -191,7 +193,9 @@ abstract class StreamerTestCase {
                 AndroidUtils.fakeValidAudioConfig(),
                 AndroidUtils.fakeValidVideoConfig()
             )
-            streamer.stopStream()
+            runBlocking {
+                streamer.stopStream()
+            }
         } catch (e: Exception) {
             Log.e(TAG, "configureStopStreamTest: exception: ", e)
             fail("Must be possible to configure/stopStream but catches exception: $e")
@@ -224,8 +228,8 @@ abstract class StreamerTestCase {
             )
             runBlocking {
                 streamer.startStream()
+                streamer.stopStream()
             }
-            streamer.stopStream()
         } catch (e: Exception) {
             Log.e(TAG, "startStreamStopStreamTest: exception: ", e)
             fail("Must be possible to startStream/stopStream but catches exception: $e")
@@ -258,8 +262,8 @@ abstract class StreamerTestCase {
             (0..10).forEach { _ ->
                 runBlocking {
                     streamer.startStream()
+                    streamer.stopStream()
                 }
-                streamer.stopStream()
             }
         } catch (e: Exception) {
             Log.e(TAG, "multipleStartStreamStopStreamTest: exception: ", e)

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/encoders/IEncoder.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/encoders/IEncoder.kt
@@ -15,9 +15,11 @@
  */
 package io.github.thibaultbee.streampack.internal.encoders
 
+import io.github.thibaultbee.streampack.internal.interfaces.Configurable
+import io.github.thibaultbee.streampack.internal.interfaces.Releaseable
 import io.github.thibaultbee.streampack.internal.interfaces.Streamable
 
-interface IEncoder<T> : Streamable<T> {
+interface IEncoder<T> : Streamable, Configurable<T>, Releaseable {
     /**
      * Input and output of an async encoder
      */

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/endpoints/FakeEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/endpoints/FakeEndpoint.kt
@@ -22,10 +22,6 @@ import io.github.thibaultbee.streampack.logger.Logger
  * A fake endpoint for test purpose.
  */
 class FakeEndpoint : IEndpoint {
-    override fun startStream() {
-        Logger.d(TAG, "startStream called")
-    }
-
     override fun configure(config: Int) {
         Logger.d(TAG, "configure called with bitrate = $config")
     }
@@ -34,7 +30,11 @@ class FakeEndpoint : IEndpoint {
         Logger.d(TAG, "write called (packet size = ${packet.buffer.remaining()})")
     }
 
-    override fun stopStream() {
+    override suspend fun startStream() {
+        Logger.d(TAG, "startStream called")
+    }
+
+    override suspend fun stopStream() {
         Logger.d(TAG, "stopStream called")
     }
 

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/endpoints/FileWriter.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/endpoints/FileWriter.kt
@@ -36,7 +36,7 @@ class FileWriter : IEndpoint {
 
     var outputStream: OutputStream? = null
 
-    override fun startStream() {
+    override suspend fun startStream() {
         if (outputStream == null) {
             throw UnsupportedOperationException("Set a file before trying to write it")
         }
@@ -49,7 +49,7 @@ class FileWriter : IEndpoint {
             ?: throw UnsupportedOperationException("Set a file before trying to write it")
     }
 
-    override fun stopStream() {
+    override suspend fun stopStream() {
         outputStream?.close()
     }
 

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/endpoints/IEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/endpoints/IEndpoint.kt
@@ -16,15 +16,11 @@
 package io.github.thibaultbee.streampack.internal.endpoints
 
 import io.github.thibaultbee.streampack.internal.data.Packet
-import io.github.thibaultbee.streampack.internal.interfaces.Streamable
+import io.github.thibaultbee.streampack.internal.interfaces.Configurable
+import io.github.thibaultbee.streampack.internal.interfaces.Releaseable
+import io.github.thibaultbee.streampack.internal.interfaces.SuspendStreamable
 
-interface IEndpoint : Streamable<Int> {
-
-    /**
-     * Configure endpoint bitrate, mainly for network endpoint.
-     * @param config bitrate at the beginning of the communication
-     */
-    override fun configure(config: Int)
+interface IEndpoint : SuspendStreamable, Configurable<Int>, Releaseable {
 
     /**
      * Writes a buffer to endpoint.

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/interfaces/Streamable.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/interfaces/Streamable.kt
@@ -15,25 +15,45 @@
  */
 package io.github.thibaultbee.streampack.internal.interfaces
 
-interface Streamable<T> {
+interface Streamable {
     /**
-     * Configure the [Streamable] implementation.
-     *
-     * @param config [Streamable] implementation configuration
+     * Starts frames or data stream generation
+     * Throws an exception if not ready for live stream
      */
-    fun configure(config: T)
+    fun startStream()
+
+    /**
+     * Stops frames or data stream generation
+     */
+    fun stopStream()
+}
 
+/**
+ * Same as [Streamable] but with suspend functions.
+ */
+interface SuspendStreamable {
     /**
      * Starts frames or data stream generation
      * Throws an exception if not ready for live stream
      */
-    fun startStream()
+    suspend fun startStream()
 
     /**
      * Stops frames or data stream generation
      */
-    fun stopStream()
+    suspend fun stopStream()
+}
+
+interface Configurable<T> {
+    /**
+     * Configure the [Configurable] implementation.
+     *
+     * @param config [Configurable] implementation configuration
+     */
+    fun configure(config: T)
+}
 
+interface Releaseable {
     /**
      * Closes and releases resources
      */

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/IMuxer.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/IMuxer.kt
@@ -18,9 +18,10 @@ package io.github.thibaultbee.streampack.internal.muxers
 import io.github.thibaultbee.streampack.data.Config
 import io.github.thibaultbee.streampack.internal.data.Frame
 import io.github.thibaultbee.streampack.internal.interfaces.ISourceOrientationProvider
+import io.github.thibaultbee.streampack.internal.interfaces.Releaseable
 import io.github.thibaultbee.streampack.internal.interfaces.Streamable
 
-interface IMuxer : Streamable<Unit> {
+interface IMuxer : Streamable, Releaseable {
     val helper: IMuxerHelper
 
     var sourceOrientationProvider: ISourceOrientationProvider?

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/flv/FlvMuxer.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/flv/FlvMuxer.kt
@@ -93,10 +93,6 @@ class FlvMuxer(
         return streamMap
     }
 
-    override fun configure(config: Unit) {
-        // Nothing to configure
-    }
-
     override fun startStream() {
         // Header
         if (writeToFile) {

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/mp4/MP4Muxer.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/mp4/MP4Muxer.kt
@@ -80,9 +80,6 @@ class MP4Muxer(
         return streamMap
     }
 
-    override fun configure(config: Unit) {
-    }
-
     override fun startStream() {
         writeBuffer(FileTypeBox().toByteBuffer())
         currentSegment = createNewSegment(MovieBoxFactory(timescale))

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/ts/TSMuxer.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/muxers/ts/TSMuxer.kt
@@ -380,10 +380,6 @@ class TSMuxer(
         service.streams.clear()
     }
 
-    override fun configure(config: Unit) {
-        // Nothing to configure
-    }
-
     override fun startStream() {
         // Nothing to start
     }

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/sources/IFrameSource.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/sources/IFrameSource.kt
@@ -16,10 +16,12 @@
 package io.github.thibaultbee.streampack.internal.sources
 
 import io.github.thibaultbee.streampack.internal.data.Frame
+import io.github.thibaultbee.streampack.internal.interfaces.Configurable
+import io.github.thibaultbee.streampack.internal.interfaces.Releaseable
 import io.github.thibaultbee.streampack.internal.interfaces.Streamable
 import java.nio.ByteBuffer
 
-interface IFrameSource<T> : Streamable<T> {
+interface IFrameSource<T> : Streamable, Configurable<T>, Releaseable {
 
     /**
      * Generate a frame from capture device

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/internal/sources/ISurfaceSource.kt b/core/src/main/java/io/github/thibaultbee/streampack/internal/sources/ISurfaceSource.kt
@@ -17,9 +17,11 @@ package io.github.thibaultbee.streampack.internal.sources
 
 import android.view.Surface
 import io.github.thibaultbee.streampack.data.VideoConfig
+import io.github.thibaultbee.streampack.internal.interfaces.Configurable
+import io.github.thibaultbee.streampack.internal.interfaces.Releaseable
 import io.github.thibaultbee.streampack.internal.interfaces.Streamable
 
-interface ISurfaceSource : Streamable<VideoConfig> {
+interface ISurfaceSource : Streamable, Configurable<VideoConfig>, Releaseable {
     /**
      * The offset between source capture time and MONOTONIC clock. It is used to synchronize video
      * with audio. It is only useful for camera source.

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/streamers/StreamerLifeCycleObserver.kt b/core/src/main/java/io/github/thibaultbee/streampack/streamers/StreamerLifeCycleObserver.kt
@@ -20,6 +20,7 @@ import androidx.lifecycle.LifecycleOwner
 import io.github.thibaultbee.streampack.streamers.interfaces.IStreamer
 import io.github.thibaultbee.streampack.utils.getCameraStreamer
 import io.github.thibaultbee.streampack.utils.getLiveStreamer
+import kotlinx.coroutines.runBlocking
 
 /**
  * Add [DefaultLifecycleObserver] to a streamer.
@@ -35,7 +36,9 @@ import io.github.thibaultbee.streampack.utils.getLiveStreamer
 open class StreamerLifeCycleObserver(var streamer: IStreamer) : DefaultLifecycleObserver {
     override fun onPause(owner: LifecycleOwner) {
         streamer.getCameraStreamer()?.stopPreview()
-        streamer.stopStream()
+        runBlocking {
+            streamer.stopStream()
+        }
         streamer.getLiveStreamer()?.let {
             if (it.isConnected) {
                 it.disconnect()

diff --git a/core/src/main/java/io/github/thibaultbee/streampack/streamers/bases/BaseCameraStreamer.kt b/core/src/main/java/io/github/thibaultbee/streampack/streamers/bases/BaseCameraStreamer.kt
@@ -117,7 +117,9 @@ open class BaseCameraStreamer(
      * @see [startPreview]
      */
     override fun stopPreview() {
-        stopStream()
+        runBlocking {
+            stopStream()
+        }
         cameraSource.stopPreview()
     }