feat(*): add metrics and regulator for RTMP

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

@@ -223,13 +223,9 @@ internal class StubAudioSyncConfigurableEncodingPipelineOutputInternal :
     override val endpoint: IEndpoint
         get() = TODO("Not yet implemented")
 
-    override fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) {
-        TODO("Not yet implemented")
-    }
-
-    override fun removeBitrateRegulatorController() {
-        TODO("Not yet implemented")
-    }
+    override var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory?
+        get() = TODO("Not yet implemented")
+        set(value) {}
 
     override suspend fun open(descriptor: MediaDescriptor) {
         TODO("Not yet implemented")
@@ -272,13 +268,9 @@ internal class StubVideoSurfaceConfigurableEncodingPipelineOutputInternal :
     override val endpoint: IEndpoint
         get() = TODO("Not yet implemented")
 
-    override fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) {
-        TODO("Not yet implemented")
-    }
-
-    override fun removeBitrateRegulatorController() {
-        TODO("Not yet implemented")
-    }
+    override var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory?
+        get() = TODO("Not yet implemented")
+        set(value) {}
 
     override suspend fun open(descriptor: MediaDescriptor) {
         TODO("Not yet implemented")

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

@@ -115,14 +115,6 @@ open class CombineEndpoint(
             .reduce { acc, iEndpointInfo -> acc intersect iEndpointInfo }
     }
 
-    /**
-     * Throws [UnsupportedOperationException] because [CombineEndpoint] does not have metrics.
-     *
-     * Call [IEndpoint.metrics] on each endpoint to get their metrics.
-     */
-    override val metrics: Any
-        get() = throw UnsupportedOperationException("CombineEndpoint does not have metrics.")
-
     private fun createNewStreamId(): Int {
         var i = 0
         while (endpointsToStreamIdsMap.keys.any { it.second == i }) {

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

@@ -41,8 +41,6 @@ class DummyEndpoint : IEndpointInternal {
         TODO("Not yet implemented")
     }
 
-    override val metrics: Any
-        get() = TODO("Not yet implemented")
     override val throwableFlow: StateFlow<Throwable?> = MutableStateFlow(null).asStateFlow()
 
     override suspend fun open(descriptor: MediaDescriptor) {

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

@@ -25,6 +25,9 @@ import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxer
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxers.ts.data.TSServiceInfo
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.ContentSink
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.FileSink
+import io.github.thibaultbee.streampack.core.elements.metrics.EmptyEndpointMetrics
+import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetrics
+import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics
 import io.github.thibaultbee.streampack.core.elements.utils.ConflatedJob
 import io.github.thibaultbee.streampack.core.logger.Logger
 import io.github.thibaultbee.streampack.core.pipelines.IDispatcherProvider
@@ -48,7 +51,7 @@ open class DynamicEndpoint(
     private val context: Context,
     private val defaultDispatcher: CoroutineDispatcher,
     private val ioDispatcher: CoroutineDispatcher
-) : IEndpointInternal {
+) : IEndpointInternal, WithEndpointMetrics<Any> {
     private val coroutineScope = CoroutineScope(defaultDispatcher)
     private val mutex = Mutex()
 
@@ -84,8 +87,8 @@ open class DynamicEndpoint(
 
     override fun getInfo(type: MediaDescriptor.Type) = getEndpoint(type).getInfo(type)
 
-    override val metrics: Any
-        get() = endpoint?.metrics ?: throw IllegalStateException("Endpoint is not opened")
+    override val metrics: EndpointMetrics<Any>
+        get() = (endpoint as? WithEndpointMetrics<Any>)?.metrics ?: EmptyEndpointMetrics
 
     init {
         coroutineScope.launch {
@@ -176,13 +179,14 @@ open class DynamicEndpoint(
     private fun prepareEndpoint(mediaDescriptor: MediaDescriptor): IEndpointInternal {
         val endpoint = getEndpoint(mediaDescriptor.type)
 
-        if (endpoint is CompositeEndpoint) {
-            if (endpoint.muxer is TsMuxer) {
+        if (endpoint is io.github.thibaultbee.streampack.core.elements.endpoints.composites.ICompositeEndpoint) {
+            val muxer = endpoint.muxer
+            if (muxer is TsMuxer) {
                 // Clean up services
-                endpoint.muxer.removeServices()
+                muxer.removeServices()
                 val serviceInfo = mediaDescriptor.getCustomData(TSServiceInfo::class.java)
                     ?: createDefaultTsServiceInfo()
-                endpoint.muxer.addService(serviceInfo)
+                muxer.addService(serviceInfo)
             }
         }
 

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

@@ -1,8 +1,9 @@
 package io.github.thibaultbee.streampack.core.elements.endpoints
 
 import android.content.Context
-import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpoint
+import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpointWithMetrics
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpoints
+import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.ISinkWithMetricsInternal
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxers.ts.TsMuxer
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxers.ts.data.TSServiceInfo
 import kotlinx.coroutines.CoroutineDispatcher
@@ -98,6 +99,6 @@ object Endpoints {
         if (serviceInfo != null) {
             muxer.addService(serviceInfo)
         }
-        return CompositeEndpoint(muxer, sink)
+        return CompositeEndpointWithMetrics(muxer, sink as ISinkWithMetricsInternal<*>)
     }
 }

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

@@ -170,9 +170,4 @@ interface IEndpoint {
             val supportedEncoders: List<String>
         }
     }
-
-    /**
-     * Metrics of the endpoint.
-     */
-    val metrics: Any
 }

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

@@ -64,9 +64,6 @@ class MediaMuxerEndpoint(
 
     override fun getInfo(type: MediaDescriptor.Type) = Companion.getInfo(type)
 
-    override val metrics: Any
-        get() = TODO("Not yet implemented")
-
     private val _isOpenFlow = MutableStateFlow(false)
     override val isOpenFlow = _isOpenFlow.asStateFlow()
 

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

@@ -25,7 +25,9 @@ import io.github.thibaultbee.streampack.core.elements.endpoints.composites.data.
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxers.IMuxer
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.muxers.IMuxerInternal
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.ISinkInternal
+import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.ISinkWithMetricsInternal
 import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.SinkConfiguration
+import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics
 import io.github.thibaultbee.streampack.core.pipelines.IDispatcherProvider
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -37,11 +39,10 @@ import kotlinx.coroutines.sync.withLock
 /**
  * An [IEndpointInternal] implementation that combines a [IMuxerInternal] and a [ISinkInternal].
  */
-class CompositeEndpoint(
+open class CompositeEndpoint(
     override val muxer: IMuxerInternal,
     override val sink: ISinkInternal
-) :
-    ICompositeEndpointInternal {
+) : ICompositeEndpointInternal {
     /**
      * The video and audio configurations.
      * It is used to configure the sink.
@@ -51,9 +52,6 @@ class CompositeEndpoint(
     override val info by lazy { EndpointInfo(muxer.info) }
     override fun getInfo(type: MediaDescriptor.Type) = info
 
-    override val metrics: Any
-        get() = sink.metrics
-
     init {
         muxer.listener = object :
             IMuxerInternal.IMuxerListener {
@@ -131,6 +129,14 @@ class CompositeEndpoint(
     }
 }
 
+/**
+ * An [IEndpointInternal] implementation of [CompositeEndpoint] with [WithEndpointMetrics].
+ */
+class CompositeEndpointWithMetrics<T : Any>(
+    muxer: IMuxerInternal,
+    sink: ISinkWithMetricsInternal<T>
+) : CompositeEndpoint(muxer, sink), WithEndpointMetrics<T> by sink
+
 /**
  * A factory to build a [CompositeEndpoint].
  */
@@ -145,3 +151,18 @@ class CompositeEndpointFactory(
         return CompositeEndpoint(muxer, sink)
     }
 }
+
+/**
+ * A factory to build a [CompositeEndpointWithMetrics].
+ */
+class CompositeEndpointWithMetricsFactory<T : Any>(
+    val muxer: IMuxerInternal,
+    val sink: ISinkWithMetricsInternal<T>
+) : IEndpointInternal.Factory {
+    override fun create(
+        context: Context,
+        dispatcherProvider: IDispatcherProvider
+    ): IEndpointInternal {
+        return CompositeEndpointWithMetrics(muxer, sink)
+    }
+}

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

@@ -13,9 +13,6 @@ import io.github.thibaultbee.streampack.core.logger.Logger
 abstract class AbstractSink : ISinkInternal {
     abstract val supportedSinkTypes: List<MediaSinkType>
 
-    override val metrics: Any
-        get() = TODO("Not yet implemented")
-
     override suspend fun open(mediaDescriptor: MediaDescriptor) {
         if (isOpenFlow.value) {
             Logger.w(TAG, "Sink is already opened")

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

@@ -22,6 +22,8 @@ import io.github.thibaultbee.streampack.core.elements.interfaces.SuspendCloseabl
 import io.github.thibaultbee.streampack.core.elements.interfaces.SuspendStreamable
 import kotlinx.coroutines.flow.StateFlow
 
+import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics
+
 interface ISinkInternal : ISink, Configurable<SinkConfiguration>, SuspendStreamable,
     SuspendCloseable {
     /**
@@ -39,15 +41,12 @@ interface ISinkInternal : ISink, Configurable<SinkConfiguration>, SuspendStreama
     suspend fun write(packet: Packet): Int
 }
 
+interface ISinkWithMetricsInternal<T : Any> : ISinkInternal, WithEndpointMetrics<T>
+
 interface ISink {
     /**
      * Whether if the endpoint is opened.
      * For example, if the file is opened for [FileSink].
      */
     val isOpenFlow: StateFlow<Boolean>
-
-    /**
-     * Metrics of the sink.
-     */
-    val metrics: Any
 }

core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/EndpointMetrics.kt

@@ -0,0 +1,127 @@
+/*
+ * Copyright (C) 2026 Thibault B.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.thibaultbee.streampack.core.elements.metrics
+
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.isActive
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
+
+interface BasicEndpointMetrics {
+    /**
+     * The duration of the interval
+     */
+    val uptime: Duration
+
+    /**
+     *  The number of packets written.
+     */
+    val packetsWritten: Long
+
+    /**
+     * The number of packets dropped before writing (e.g. due to congestion or timeout).
+     */
+    val packetsWriteDropped: Long
+
+    /**
+     * The number of packets lost during the transmission.
+     */
+    val packetsWriteLost: Long
+
+    /**
+     * The number of bytes successfully written.
+     */
+    val bytesWritten: Long
+
+    /**
+     * The number of bytes dropped before writing (e.g. due to congestion or timeout).
+     */
+    val bytesWriteDropped: Long
+    /**
+     * Subtracts two [BasicEndpointMetrics]s.
+     */
+    operator fun minus(other: BasicEndpointMetrics): BasicEndpointMetrics {
+        return object : BasicEndpointMetrics {
+            override val uptime = this@BasicEndpointMetrics.uptime - other.uptime
+            override val packetsWritten = this@BasicEndpointMetrics.packetsWritten - other.packetsWritten
+            override val packetsWriteDropped = this@BasicEndpointMetrics.packetsWriteDropped - other.packetsWriteDropped
+            override val packetsWriteLost = this@BasicEndpointMetrics.packetsWriteLost - other.packetsWriteLost
+            override val bytesWritten = this@BasicEndpointMetrics.bytesWritten - other.bytesWritten
+            override val bytesWriteDropped = this@BasicEndpointMetrics.bytesWriteDropped - other.bytesWriteDropped
+        }
+    }
+}
+
+/**
+ * The total written bitrate in bits per second (bps).
+ */
+val BasicEndpointMetrics.writtenBitrateInBps: Long
+    get() = uptime.inWholeMilliseconds.let { if (it == 0L) 0L else (bytesWritten * 8000) / it }
+
+/**
+ * Endpoint metrics interface
+ */
+interface EndpointMetrics<out T : Any> : BasicEndpointMetrics {
+    /**
+     * The implementation-specific metrics wrapper.
+     */
+    val rawMetrics: T
+}
+
+/**
+ * An empty implementation of [EndpointMetrics] that returns zeros for all metrics.
+ */
+object EmptyEndpointMetrics : EndpointMetrics<Any> {
+    override val uptime: Duration = Duration.ZERO
+    override val packetsWritten: Long = 0L
+    override val packetsWriteDropped: Long = 0L
+    override val packetsWriteLost: Long = 0L
+    override val bytesWritten: Long = 0L
+    override val bytesWriteDropped: Long = 0L
+    override val rawMetrics: Any = Unit
+}
+
+/**
+ * A specific [WithMetrics] for [EndpointMetrics].
+ *
+ * The members from [BasicEndpointMetrics] represent cumulative metrics.
+ */
+interface WithEndpointMetrics<out T : Any> : WithMetrics<EndpointMetrics<T>>
+
+/**
+ * Represents a pair of instant and cumulative metrics.
+ */
+data class TrackedMetrics(
+    val instant: BasicEndpointMetrics,
+    val cumulative: BasicEndpointMetrics
+)
+
+/**
+ * Returns a Flow that emits the [BasicEndpointMetrics] difference since the last emission.
+ * Every collector gets its own isolated [EndpointMetricsTracker] to prevent state collisions.
+ *
+ * @param interval The delay between emissions.
+ */
+fun WithEndpointMetrics<*>.metricsFlow(interval: Duration = 1000.milliseconds): Flow<TrackedMetrics> = flow {
+    val tracker = EndpointMetricsTracker(this@metricsFlow)
+    while (currentCoroutineContext().isActive) {
+        delay(interval)
+        emit(TrackedMetrics(tracker.instant, tracker.cumulative))
+    }
+}