refactor(core): endpoint: pass the frame closeable in the frame parameter passed for IEndpointInternal.write

Author: ThibaultBee

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

@@ -19,6 +19,7 @@ import android.content.Context
 import android.util.Log
 import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.MediaDescriptor
 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.pipelines.IDispatcherProvider
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -63,14 +64,15 @@ class DummyEndpoint : IEndpointInternal {
         _isOpenFlow.emit(false)
     }
 
-    override suspend fun write(frame: Frame, streamPid: Int, onFrameProcessed: (() -> Unit)) {
+    override suspend fun write(closeableFrame: FrameWithCloseable, streamPid: Int) {
+        val frame = closeableFrame.frame
         Log.i(TAG, "write: $frame")
         _frameFlow.emit(frame)
         when {
             frame.isAudio -> numOfAudioFramesWritten++
             frame.isVideo -> numOfVideoFramesWritten++
         }
-        onFrameProcessed()
+        closeableFrame.close()
     }
 
     override fun addStreams(streamConfigs: List<CodecConfig>): Map<CodecConfig, Int> {

core/src/main/java/io/github/thibaultbee/streampack/core/elements/data/Frame.kt

@@ -148,8 +148,8 @@ fun CloseableFrame(
     /**
      * A callback to call when frame is closed.
      */
-    onClosed: (CloseableFrame) -> Unit,
-) = CloseableFrame(
+    onClosed: (FrameWithCloseable) -> Unit,
+) = FrameWithCloseable(
     Frame(
         rawBuffer,
         ptsInUs,
@@ -163,9 +163,9 @@ fun CloseableFrame(
 /**
  * Frame internal representation.
  */
-data class CloseableFrame(
+data class FrameWithCloseable(
     val frame: Frame,
-    val onClosed: (CloseableFrame) -> Unit
+    val onClosed: (FrameWithCloseable) -> Unit
 ) : Closeable {
     override fun close() {
         try {

core/src/main/java/io/github/thibaultbee/streampack/core/elements/encoders/IEncoder.kt

@@ -16,7 +16,7 @@
 package io.github.thibaultbee.streampack.core.elements.encoders
 
 import android.view.Surface
-import io.github.thibaultbee.streampack.core.elements.data.CloseableFrame
+import io.github.thibaultbee.streampack.core.elements.data.FrameWithCloseable
 import io.github.thibaultbee.streampack.core.elements.data.RawFrame
 import io.github.thibaultbee.streampack.core.elements.interfaces.Releasable
 import io.github.thibaultbee.streampack.core.elements.interfaces.SuspendStreamable
@@ -77,7 +77,7 @@ interface IEncoderInternal : SuspendStreamable, Releasable,
         /**
          * A channel where the encoder will send encoded frames.
          */
-        val outputChannel: SendChannel<CloseableFrame>
+        val outputChannel: SendChannel<FrameWithCloseable>
     }
 
     /**

core/src/main/java/io/github/thibaultbee/streampack/core/elements/encoders/mediacodec/MediaCodecEncoder.kt

@@ -23,6 +23,7 @@ import android.os.Bundle
 import android.util.Log
 import android.view.Surface
 import io.github.thibaultbee.streampack.core.elements.data.CloseableFrame
+import io.github.thibaultbee.streampack.core.elements.data.FrameWithCloseable
 import io.github.thibaultbee.streampack.core.elements.data.RawFrame
 import io.github.thibaultbee.streampack.core.elements.encoders.EncoderMode
 import io.github.thibaultbee.streampack.core.elements.encoders.IEncoderInternal
@@ -593,7 +594,7 @@ internal constructor(
          */
         fun frame(
             index: Int, outputFormat: MediaFormat, info: BufferInfo, tag: String
-        ): CloseableFrame {
+        ): FrameWithCloseable {
             var pts = info.presentationTimeUs
             if (pts <= previousPresentationTimestamp) {
                 pts = previousPresentationTimestamp + 1
@@ -618,7 +619,7 @@ internal constructor(
          */
         private fun Frame(
             codec: MediaCodec, index: Int, outputFormat: MediaFormat, info: BufferInfo, tag: String
-        ): CloseableFrame {
+        ): FrameWithCloseable {
             val buffer = requireNotNull(codec.getOutputBuffer(index))
             return CloseableFrame(
                 buffer, info.presentationTimeUs, // pts

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

@@ -17,7 +17,7 @@ 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.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.utils.extensions.intersect
 import io.github.thibaultbee.streampack.core.logger.Logger
@@ -209,7 +209,8 @@ open class CombineEndpoint(
      *
      * If all endpoints write fails, it throws the exception of the first endpoint that failed.
      */
-    override suspend fun write(frame: Frame, streamPid: Int, onFrameProcessed: (() -> Unit)) {
+    override suspend fun write(closeableFrame: FrameWithCloseable, streamPid: Int) {
+        val frame = closeableFrame.frame
         val throwables = mutableListOf<Throwable>()
 
         /**
@@ -221,11 +222,13 @@ open class CombineEndpoint(
         endpointInternals.filter { it.isOpenFlow.value }.forEach { endpoint ->
             try {
                 val deferred = CompletableDeferred<Unit>()
-                val duplicatedFrame = frame.copy(rawBuffer = frame.rawBuffer.duplicate())
-                val endpointStreamId = endpointsToStreamIdsMap[Pair(endpoint, streamPid)]!!
+                val duplicatedFrame = FrameWithCloseable(
+                    frame.copy(rawBuffer = frame.rawBuffer.duplicate()),
+                    { deferred.complete(Unit) })
 
+                val endpointStreamId = endpointsToStreamIdsMap[Pair(endpoint, streamPid)]!!
                 deferreds += deferred
-                endpoint.write(duplicatedFrame, endpointStreamId, { deferred.complete(Unit) })
+                endpoint.write(duplicatedFrame, endpointStreamId)
             } catch (t: Throwable) {
                 Logger.e(TAG, "Failed to get stream id for endpoint $endpoint", t)
                 throwables += t
@@ -234,7 +237,7 @@ open class CombineEndpoint(
 
         coroutineScope.launch {
             deferreds.forEach { it.await() }
-            onFrameProcessed()
+            closeableFrame.close()
         }
 
         if (throwables.isNotEmpty()) {

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

@@ -18,7 +18,7 @@ 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.configuration.mediadescriptor.createDefaultTsServiceInfo
-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.CompositeEndpoint
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpoints
@@ -131,8 +131,8 @@ open class DynamicEndpoint(
         }
     }
 
-    override suspend fun write(frame: Frame, streamPid: Int, onFrameProcessed: () -> Unit) =
-        safeEndpoint { endpoint -> endpoint.write(frame, streamPid, onFrameProcessed) }
+    override suspend fun write(closeableFrame: FrameWithCloseable, streamPid: Int) =
+        safeEndpoint { endpoint -> endpoint.write(closeableFrame, streamPid) }
 
     override suspend fun startStream() = safeEndpoint { endpoint -> endpoint.startStream() }
 

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

@@ -17,6 +17,7 @@ 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.encoders.CodecConfig
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.FileSink
@@ -39,18 +40,15 @@ interface IEndpointInternal : IEndpoint, SuspendStreamable,
     /**
      * Writes a [Frame] to the [IEndpointInternal].
      *
-     * The [onFrameProcessed] callback must be called when the frame has been processed and the [Frame.rawBuffer] is not used anymore.
-     * The [IEndpointInternal] must called [onFrameProcessed] even if the frame is dropped or it somehow crashes.
-     * Also, once [onFrameProcessed] is called, the [Frame.rawBuffer] must not be used anymore by the [IEndpointInternal].
+     * The [FrameWithCloseable.close] must be called when the frame has been processed and the [Frame.rawBuffer] is not used anymore.
+     * The [IEndpointInternal] must called [FrameWithCloseable.close] even if the frame is dropped or it somehow crashes.
      *
-     * @param frame the [Frame] to write
+     * @param closeableFrame the [Frame] to write
      * @param streamPid the stream id the [Frame] belongs to
-     * @param onFrameProcessed a callback called when the [Frame.rawBuffer] is not used anymore
      */
     suspend fun write(
-        frame: Frame,
+        closeableFrame: FrameWithCloseable,
         streamPid: Int,
-        onFrameProcessed: (() -> Unit)
     )
 
     /**

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

@@ -24,7 +24,7 @@ import android.media.MediaMuxer.OutputFormat
 import android.os.Build
 import android.os.ParcelFileDescriptor
 import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.MediaDescriptor
-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.logger.Logger
 import io.github.thibaultbee.streampack.core.pipelines.IDispatcherProvider
@@ -140,8 +140,9 @@ class MediaMuxerEndpoint(
     }
 
     override suspend fun write(
-        frame: Frame, streamPid: Int, onFrameProcessed: () -> Unit
+        closeableFrame: FrameWithCloseable, streamPid: Int
     ) = withContext(ioDispatcher) {
+        val frame = closeableFrame.frame
         mutex.withLock {
             try {
                 if (state != State.STARTED && state != State.PENDING_START) {
@@ -180,7 +181,7 @@ class MediaMuxerEndpoint(
                 Logger.e(TAG, "Error while writing frame: ${t.message}")
                 throw t
             } finally {
-                onFrameProcessed()
+                closeableFrame.close()
             }
         }
     }

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

@@ -17,7 +17,7 @@ package io.github.thibaultbee.streampack.core.elements.endpoints.composites
 
 import android.content.Context
 import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.MediaDescriptor
-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.data.Packet
 import io.github.thibaultbee.streampack.core.elements.encoders.CodecConfig
 import io.github.thibaultbee.streampack.core.elements.endpoints.IEndpoint
@@ -75,10 +75,9 @@ class CompositeEndpoint(
     }
 
     override suspend fun write(
-        frame: Frame,
-        streamPid: Int,
-        onFrameProcessed: (() -> Unit)
-    ) = muxer.write(frame, streamPid, onFrameProcessed)
+        closeableFrame: FrameWithCloseable,
+        streamPid: Int
+    ) = muxer.write(closeableFrame, streamPid)
 
     override fun addStreams(streamConfigs: List<CodecConfig>): Map<CodecConfig, Int> {
         mutex.tryLock()

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

@@ -15,7 +15,7 @@
  */
 package io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxers
 
-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.data.Packet
 import io.github.thibaultbee.streampack.core.elements.encoders.CodecConfig
 import io.github.thibaultbee.streampack.core.elements.interfaces.Releasable
@@ -32,7 +32,7 @@ interface IMuxerInternal :
         fun onOutputFrame(packet: Packet)
     }
 
-    fun write(frame: Frame, streamPid: Int, onFrameProcessed: () -> Unit)
+    fun write(closeableFrame: FrameWithCloseable, streamPid: Int)
 
     fun addStreams(streamsConfig: List<CodecConfig>): Map<CodecConfig, Int>