diff --git a/core/src/androidTest/java/io/github/thibaultbee/streampack/core/pipelines/outputs/StubPipelineOutput.kt b/core/src/androidTest/java/io/github/thibaultbee/streampack/core/pipelines/outputs/StubPipelineOutput.kt index f133bf2a5..6ae4db862 100644 --- a/core/src/androidTest/java/io/github/thibaultbee/streampack/core/pipelines/outputs/StubPipelineOutput.kt +++ b/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") diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/CombineEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/CombineEndpoint.kt index f9d235323..190e95b7e 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/CombineEndpoint.kt +++ b/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 }) { diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DummyEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DummyEndpoint.kt index 429b0a078..9d5b06d0a 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DummyEndpoint.kt +++ b/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 = MutableStateFlow(null).asStateFlow() override suspend fun open(descriptor: MediaDescriptor) { diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DynamicEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DynamicEndpoint.kt index 1951fabc0..a9fc04263 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/DynamicEndpoint.kt +++ b/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 { 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 + get() = (endpoint as? WithEndpointMetrics)?.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) } } diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/Endpoints.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/Endpoints.kt index e31a2710d..f9353a32e 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/Endpoints.kt +++ b/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<*>) } } diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/IEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/IEndpoint.kt index 715c7d5c8..8f4fa80be 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/IEndpoint.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/IEndpoint.kt @@ -170,9 +170,4 @@ interface IEndpoint { val supportedEncoders: List } } - - /** - * Metrics of the endpoint. - */ - val metrics: Any } \ No newline at end of file diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/MediaMuxerEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/MediaMuxerEndpoint.kt index cedb2462a..0be8a776e 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/MediaMuxerEndpoint.kt +++ b/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() diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/CompositeEndpoint.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/CompositeEndpoint.kt index e85726a65..51493a843 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/CompositeEndpoint.kt +++ b/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( + muxer: IMuxerInternal, + sink: ISinkWithMetricsInternal +) : CompositeEndpoint(muxer, sink), WithEndpointMetrics 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( + val muxer: IMuxerInternal, + val sink: ISinkWithMetricsInternal +) : IEndpointInternal.Factory { + override fun create( + context: Context, + dispatcherProvider: IDispatcherProvider + ): IEndpointInternal { + return CompositeEndpointWithMetrics(muxer, sink) + } +} diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/sinks/AbstractSink.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/sinks/AbstractSink.kt index 91ac63a37..de4c45efa 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/sinks/AbstractSink.kt +++ b/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 - 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") diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/sinks/ISink.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/sinks/ISink.kt index 1e36c9cdd..052817cfe 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/endpoints/composites/sinks/ISink.kt +++ b/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, SuspendStreamable, SuspendCloseable { /** @@ -39,15 +41,12 @@ interface ISinkInternal : ISink, Configurable, SuspendStreama suspend fun write(packet: Packet): Int } +interface ISinkWithMetricsInternal : ISinkInternal, WithEndpointMetrics + interface ISink { /** * Whether if the endpoint is opened. * For example, if the file is opened for [FileSink]. */ val isOpenFlow: StateFlow - - /** - * Metrics of the sink. - */ - val metrics: Any } \ No newline at end of file diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/EndpointMetrics.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/EndpointMetrics.kt new file mode 100644 index 000000000..1aded53d6 --- /dev/null +++ b/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 : BasicEndpointMetrics { + /** + * The implementation-specific metrics wrapper. + */ + val rawMetrics: T +} + +/** + * An empty implementation of [EndpointMetrics] that returns zeros for all metrics. + */ +object EmptyEndpointMetrics : EndpointMetrics { + 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 : WithMetrics> + +/** + * 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 = flow { + val tracker = EndpointMetricsTracker(this@metricsFlow) + while (currentCoroutineContext().isActive) { + delay(interval) + emit(TrackedMetrics(tracker.instant, tracker.cumulative)) + } +} \ No newline at end of file diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/EndpointMetricsTracker.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/EndpointMetricsTracker.kt new file mode 100644 index 000000000..52e8cdc29 --- /dev/null +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/EndpointMetricsTracker.kt @@ -0,0 +1,59 @@ +/* + * 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 + +/** + * Tracks [BasicEndpointMetrics] over time, providing both cumulative and instant (diff since last read) metrics. + */ +class EndpointMetricsTracker internal constructor(private val metricsProvider: WithEndpointMetrics<*>) { + private var lastMetrics: BasicEndpointMetrics? = null + + /** + * The cumulative metrics since the start. + */ + val cumulative: BasicEndpointMetrics + get() = metricsProvider.metrics + + /** + * The instant metrics (difference between the current and the previous read). + * If no previous read exists, it equals the current metrics. + * Avoid calling this from 2 different parts. + */ + @get:Synchronized + val instant: BasicEndpointMetrics + get() { + val current = metricsProvider.metrics + val last = lastMetrics + lastMetrics = current + + return if (last != null) { + current - last + } else { + current + } + } + + /** + * The implementation-specific metrics wrapper. + */ + val rawMetrics + get() = metricsProvider.metrics.rawMetrics +} + +/** + * Creates an [EndpointMetricsTracker] for this [WithEndpointMetrics]. + */ +fun WithEndpointMetrics<*>.createMetricsTracker() = EndpointMetricsTracker(this) diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/WithMetrics.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/WithMetrics.kt new file mode 100644 index 000000000..e0bff0cbb --- /dev/null +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/metrics/WithMetrics.kt @@ -0,0 +1,26 @@ +/* + * 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 + +/** + * Minimal metrics interface + */ +interface WithMetrics { + /** + * Metrics of the element. + */ + val metrics: T +} diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/sources/video/camera/CameraSettings.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/sources/video/camera/CameraSettings.kt index a1392623a..c6c85a4c1 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/sources/video/camera/CameraSettings.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/sources/video/camera/CameraSettings.kt @@ -56,7 +56,7 @@ import io.github.thibaultbee.streampack.core.elements.sources.video.camera.exten import io.github.thibaultbee.streampack.core.elements.sources.video.camera.extensions.sensitivityRange import io.github.thibaultbee.streampack.core.elements.sources.video.camera.extensions.zoomRatioRange import io.github.thibaultbee.streampack.core.elements.sources.video.camera.utils.CaptureResultListener -import io.github.thibaultbee.streampack.core.elements.utils.extensions.clamp +import io.github.thibaultbee.streampack.core.elements.utils.extensions.coerceIn import io.github.thibaultbee.streampack.core.elements.utils.extensions.isApplicationPortrait import io.github.thibaultbee.streampack.core.elements.utils.extensions.isNormalized import io.github.thibaultbee.streampack.core.elements.utils.extensions.launchIn @@ -533,7 +533,7 @@ class CameraSettings internal constructor( suspend fun setSensorSensitivity(sensorSensitivity: Int) { cameraSettings.set( CaptureRequest.SENSOR_SENSITIVITY, - sensorSensitivity.clamp(availableSensorSensitivityRange) + sensorSensitivity.coerceIn(availableSensorSensitivityRange) ) cameraSettings.applyRepeatingSession() } @@ -693,7 +693,7 @@ class CameraSettings internal constructor( suspend fun setCompensation(compensation: Int) { cameraSettings.set( CaptureRequest.CONTROL_AE_EXPOSURE_COMPENSATION, - compensation.clamp(availableCompensationRange) + compensation.coerceIn(availableCompensationRange) ) cameraSettings.applyRepeatingSession() } @@ -777,7 +777,7 @@ class CameraSettings internal constructor( suspend fun onPinch(scale: Float) { val scaledRatio: Float = getZoomRatio() * speedUpZoomByX(scale, 2) // Clamp the ratio with the zoom range. - setZoomRatio(scaledRatio.clamp(availableRatioRange.lower, availableRatioRange.upper)) + setZoomRatio(scaledRatio.coerceIn(availableRatioRange.lower, availableRatioRange.upper)) } private fun speedUpZoomByX(scaleFactor: Float, ratio: Int): Float { @@ -825,7 +825,7 @@ class CameraSettings internal constructor( @RequiresPermission(Manifest.permission.CAMERA) override suspend fun setZoomRatio(zoomRatio: Float) { mutex.withLock { - val clampedValue = zoomRatio.clamp(availableRatioRange) + val clampedValue = zoomRatio.coerceIn(availableRatioRange) if (clampedValue == persistentZoomRatio) { return@withLock } @@ -905,7 +905,7 @@ class CameraSettings internal constructor( return } cameraSettings.set( - CaptureRequest.CONTROL_ZOOM_RATIO, zoomRatio.clamp(availableRatioRange) + CaptureRequest.CONTROL_ZOOM_RATIO, zoomRatio.coerceIn(availableRatioRange) ) cameraSettings.applyRepeatingSession() notifyZoomListeners(zoomRatio) @@ -1021,7 +1021,7 @@ class CameraSettings internal constructor( @RequiresPermission(Manifest.permission.CAMERA) suspend fun setLensDistance(lensDistance: Float) { cameraSettings.set( - CaptureRequest.LENS_FOCUS_DISTANCE, lensDistance.clamp(availableLensDistanceRange) + CaptureRequest.LENS_FOCUS_DISTANCE, lensDistance.coerceIn(availableLensDistanceRange) ) cameraSettings.applyRepeatingSession() } @@ -1618,10 +1618,10 @@ class CameraSettings internal constructor( (centerY + height / 2).toInt() ) - focusRect.left = focusRect.left.clamp(cropRegion.right, cropRegion.left) - focusRect.right = focusRect.right.clamp(cropRegion.right, cropRegion.left) - focusRect.top = focusRect.top.clamp(cropRegion.bottom, cropRegion.top) - focusRect.bottom = focusRect.bottom.clamp(cropRegion.bottom, cropRegion.top) + focusRect.left = focusRect.left.coerceIn(cropRegion.right, cropRegion.left) + focusRect.right = focusRect.right.coerceIn(cropRegion.right, cropRegion.left) + focusRect.top = focusRect.top.coerceIn(cropRegion.bottom, cropRegion.top) + focusRect.bottom = focusRect.bottom.coerceIn(cropRegion.bottom, cropRegion.top) return MeteringRectangle(focusRect, DEFAULT_METERING_WEIGHT_MAX) } diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/ChannelWithCloseableData.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/ChannelWithCloseableData.kt index 130cf4450..35a803271 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/ChannelWithCloseableData.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/ChannelWithCloseableData.kt @@ -34,10 +34,14 @@ import java.io.Closeable */ class ChannelWithCloseableData( capacity: Int = RENDEZVOUS, - onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND + onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND, + onUndeliveredElement: ((T) -> Unit) = {} ) : ReceiveChannel> { private val channel = - Channel>(capacity, onBufferOverflow, onUndeliveredElement = { it.close() }) + Channel>(capacity, onBufferOverflow, onUndeliveredElement = { + it.close() + onUndeliveredElement(it.data) + }) /** * Sends data along with a close action to the channel. diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/CoroutineScheduler.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/CoroutineScheduler.kt index 2c4e88c51..250349654 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/CoroutineScheduler.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/CoroutineScheduler.kt @@ -22,8 +22,10 @@ import kotlinx.coroutines.cancel import kotlinx.coroutines.delay import kotlinx.coroutines.launch +import kotlin.time.Duration + class CoroutineScheduler( - private val delayTimeInMs: Long, + private val delayTime: Duration, coroutineDispatcher: CoroutineDispatcher, private val action: suspend CoroutineScope.() -> Unit ) { @@ -36,7 +38,7 @@ class CoroutineScheduler( } job = coroutineScope.launch { while (true) { - delay(delayTimeInMs) + delay(delayTime) launch { action() } } } diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/extensions/Extensions.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/extensions/Extensions.kt index 5d5fba851..9f59bc79a 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/extensions/Extensions.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/elements/utils/extensions/Extensions.kt @@ -43,16 +43,16 @@ internal fun Any.numOfBits(): Int { } } -internal fun > T.clamp(min: T, max: T): T { - return if (max >= min) { - if (this < min) min else if (this > max) max else this - } else { - if (this < max) max else if (this > min) min else this +internal fun > T.coerceIn(min: T, max: T): T { + return when { + this < min -> min + this > max -> max + else -> this } } -internal fun > T.clamp(range: Range) = - this.clamp(range.lower, range.upper) +internal fun > T.coerceIn(range: Range) = + this.coerceIn(range.lower, range.upper) internal val PointF.isNormalized: Boolean get() = x in 0f..1f && y in 0f..1f diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/EncodingPipelineOutput.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/EncodingPipelineOutput.kt index dcd06bc12..e6749ee68 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/EncodingPipelineOutput.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/EncodingPipelineOutput.kt @@ -102,6 +102,28 @@ internal class EncodingPipelineOutput( private val isReleaseRequested = AtomicBoolean(false) private var bitrateRegulatorController: IBitrateRegulatorController? = null + override var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory? = null + set(value) { + field = value + if (isReleaseRequested.get()) { + throw IllegalStateException("Output is released") + } + bitrateRegulatorController?.stop() + bitrateRegulatorController = + value?.newBitrateRegulatorController(this, dispatcherProvider.default) + + if (bitrateRegulatorController != null) { + Logger.d( + TAG, + "Bitrate regulator controller added: ${bitrateRegulatorController!!.javaClass.simpleName}" + ) + if (isStreaming) { + bitrateRegulatorController?.start() + } + } else { + Logger.d(TAG, "Bitrate regulator controller removed") + } + } private var audioStreamId: Int? = null private var videoStreamId: Int? = null @@ -257,6 +279,16 @@ internal class EncodingPipelineOutput( } init { + coroutineScope.launch { + isStreamingFlow.collect { isStreaming -> + if (isStreaming) { + bitrateRegulatorController?.start() + } else { + bitrateRegulatorController?.stop() + } + } + } + if (withAudio) { coroutineScope.launch(audioOutputDispatcher) { // Audio @@ -637,8 +669,6 @@ internal class EncodingPipelineOutput( videoEncoderJob?.join() endpointInternal.startStream() - - bitrateRegulatorController?.start() } catch (t: Throwable) { stopStreamUnsafe() throw t @@ -716,12 +746,6 @@ internal class EncodingPipelineOutput( * @see [stopStream] */ private suspend fun stopStreamElements() { - try { - bitrateRegulatorController?.stop() - } catch (t: Throwable) { - Logger.w(TAG, "Can't stop bitrate regulator controller: ${t.message}") - } - // Encoders val audioEncoderJob = audioEncoderInternal?.let { coroutineScope.launch { @@ -826,40 +850,6 @@ internal class EncodingPipelineOutput( coroutineScope.cancel() } - /** - * Adds a bitrate regulator controller. - * - * Limitation: it is only available for SRT for now. - */ - override fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) { - if (isReleaseRequested.get()) { - throw IllegalStateException("Output is released") - } - bitrateRegulatorController?.stop() - bitrateRegulatorController = - controllerFactory.newBitrateRegulatorController(this, dispatcherProvider.default) - .apply { - if (isStreaming) { - this.start() - } - Logger.d( - TAG, "Bitrate regulator controller added: ${this.javaClass.simpleName}" - ) - } - } - - /** - * Removes the bitrate regulator controller. - */ - override fun removeBitrateRegulatorController() { - if (isReleaseRequested.get()) { - throw IllegalStateException("Output is released") - } - bitrateRegulatorController?.stop() - bitrateRegulatorController = null - Logger.d(TAG, "Bitrate regulator controller removed") - } - private suspend fun setTargetRotationInternal(@RotationValue newTargetRotation: Int) { if (shouldUpdateRotation(newTargetRotation)) { updateVideoEncoderForTransformation() diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/IEncodingPipelineOutput.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/IEncodingPipelineOutput.kt index 9c8b3f247..5af6c9bb9 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/IEncodingPipelineOutput.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/pipelines/outputs/encoding/IEncodingPipelineOutput.kt @@ -42,12 +42,7 @@ interface IEncodingPipelineOutput : IPipelineOutput, IOpenableStreamer { /** * Adds a bitrate regulator controller to the streamer. */ - fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) - - /** - * Removes the bitrate regulator controller from the streamer. - */ - fun removeBitrateRegulatorController() + var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory? } /** @@ -120,4 +115,18 @@ interface IConfigurableAudioVideoEncodingPipelineOutput : internal interface IEncodingPipelineOutputInternal : IConfigurableAudioVideoEncodingPipelineOutput, IConfigurableAudioPipelineOutputInternal, IConfigurableVideoPipelineOutputInternal, - IPipelineEventOutputInternal \ No newline at end of file + IPipelineEventOutputInternal + +/** + * Adds a bitrate regulator controller to the streamer. + */ +fun IEncodingPipelineOutput.addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) { + this.bitrateRegulatorControllerFactory = controllerFactory +} + +/** + * Removes the bitrate regulator controller from the streamer. + */ +fun IEncodingPipelineOutput.removeBitrateRegulatorController() { + this.bitrateRegulatorControllerFactory = null +} \ No newline at end of file diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/BitrateRegulator.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/BitrateRegulator.kt index ba7c86fb4..1edb6a28c 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/BitrateRegulator.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/BitrateRegulator.kt @@ -16,6 +16,8 @@ package io.github.thibaultbee.streampack.core.regulator import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetricsTracker +import io.github.thibaultbee.streampack.core.elements.utils.extensions.coerceIn /** * Abstract class for the bitrate regulation implementation. @@ -23,12 +25,23 @@ import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfi * If you want to implement your custom bitrate regulator, it must inherit from this class. * The bitrate regulator object is created by streamers through the [IBitrateRegulator.Factory]. * + * @param metricsTracker endpoint metrics tracker * @param bitrateRegulatorConfig bitrate regulation configuration * @param onVideoTargetBitrateChange call when you have to change video bitrate * @param onAudioTargetBitrateChange call when you have to change audio bitrate */ abstract class BitrateRegulator( + override val metricsTracker: EndpointMetricsTracker, protected val bitrateRegulatorConfig: BitrateRegulatorConfig, - protected val onVideoTargetBitrateChange: ((Int) -> Unit), - protected val onAudioTargetBitrateChange: ((Int) -> Unit) -) : IBitrateRegulator + onVideoTargetBitrateChange: ((Int) -> Unit), + onAudioTargetBitrateChange: ((Int) -> Unit) +) : IBitrateRegulator { + + protected val onVideoTargetBitrateChange: ((Int) -> Unit) = { + onVideoTargetBitrateChange(it.coerceIn(bitrateRegulatorConfig.videoBitrateRange)) + } + + protected val onAudioTargetBitrateChange: ((Int) -> Unit) = { + onAudioTargetBitrateChange(it.coerceIn(bitrateRegulatorConfig.audioBitrateRange)) + } +} diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/IBitrateRegulator.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/IBitrateRegulator.kt index 367af6973..2285505ec 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/IBitrateRegulator.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/IBitrateRegulator.kt @@ -16,19 +16,24 @@ package io.github.thibaultbee.streampack.core.regulator import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetricsTracker /** * Interface to implement a bitrate regulator. */ interface IBitrateRegulator { /** - * Calls regularly to get new stats + * Tracker for endpoint metrics + */ + val metricsTracker: EndpointMetricsTracker + + /** + * Called regularly to get new metrics * - * @param stats transmission stats * @param currentVideoBitrate current video bitrate target in bits/s. * @param currentAudioBitrate current audio bitrate target in bits/s. */ - fun update(stats: Any, currentVideoBitrate: Int, currentAudioBitrate: Int) + fun update(currentVideoBitrate: Int, currentAudioBitrate: Int) /** * Factory interface you must use to create a [BitrateRegulator] object. @@ -39,12 +44,14 @@ interface IBitrateRegulator { /** * Creates a [BitrateRegulator] object from given parameters * + * @param metricsTracker endpoint metrics tracker * @param bitrateRegulatorConfig bitrate regulation configuration * @param onVideoTargetBitrateChange call when you have to change video bitrate * @param onAudioTargetBitrateChange call when you have to change audio bitrate * @return a [BitrateRegulator] object */ fun newBitrateRegulator( + metricsTracker: EndpointMetricsTracker, bitrateRegulatorConfig: BitrateRegulatorConfig, onVideoTargetBitrateChange: ((Int) -> Unit), onAudioTargetBitrateChange: ((Int) -> Unit) diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/SimpleBitrateRegulator.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/SimpleBitrateRegulator.kt new file mode 100644 index 000000000..58108a39d --- /dev/null +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/SimpleBitrateRegulator.kt @@ -0,0 +1,107 @@ +package io.github.thibaultbee.streampack.core.regulator + +import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetricsTracker +import io.github.thibaultbee.streampack.core.elements.metrics.writtenBitrateInBps +import kotlin.math.max +import kotlin.math.min + +/** + * A [BitrateRegulator] that reduce video bitrate when packets are lost. + * + * @param bitrateRegulatorConfig bitrate regulation configuration + * @param onVideoTargetBitrateChange call when you have to change video bitrate + * @param onAudioTargetBitrateChange call when you have to change audio bitrate + */ +class SimpleBitrateRegulator( + metricsTracker: EndpointMetricsTracker, + bitrateRegulatorConfig: BitrateRegulatorConfig, + onVideoTargetBitrateChange: ((Int) -> Unit), + onAudioTargetBitrateChange: ((Int) -> Unit) +) : BitrateRegulator( + metricsTracker, + bitrateRegulatorConfig, + onVideoTargetBitrateChange, + onAudioTargetBitrateChange +) { + companion object { + private const val MIN_DECREASE_STEP = 100000 // b/s + private const val MAX_INCREASE_STEP = 200000 // b/s + + private const val MAX_PERCENTAGE_DECREASE = 85 // % + private const val MIN_PERCENTAGE_DECREASE = 20 // % + } + + /** + * Called regularly to get new endpoint metrics + * + * @param currentVideoBitrate current video bitrate target in bits/s. + * @param currentAudioBitrate current audio bitrate target in bits/s. + */ + override fun update( + currentVideoBitrate: Int, + currentAudioBitrate: Int + ) { + val metrics = metricsTracker.instant + val packetsLostOrDropped = metrics.packetsWriteDropped + metrics.packetsWriteLost + val writtenBitrate = metrics.writtenBitrateInBps + + if (packetsLostOrDropped > 0) { + // Detected packet dropped or loss - we should reduce the bitrate with multiplicative decrease + // How critical? + val percentageReduction = if (metrics.packetsWritten == 0L) { + MAX_PERCENTAGE_DECREASE + } else { + (packetsLostOrDropped * 100 / metrics.packetsWritten).toInt().coerceIn(MIN_PERCENTAGE_DECREASE, MAX_PERCENTAGE_DECREASE) + } + + // Reduce current bitrate by percentageReduction % + val newVideoBitrate = currentVideoBitrate - max( + currentVideoBitrate * percentageReduction / 100, + MIN_DECREASE_STEP // getting down by 100000 b/s minimum + ) + onVideoTargetBitrateChange(newVideoBitrate) + } else if (currentVideoBitrate < bitrateRegulatorConfig.videoBitrateRange.upper) { + // Only increase if we are successfully sending at near the current target bitrate + // This prevents us from increasing the target when the network is already saturated + if (writtenBitrate > currentVideoBitrate * 0.9) { + // Additive increase + val newVideoBitrate = min( + currentVideoBitrate + MAX_INCREASE_STEP, + bitrateRegulatorConfig.videoBitrateRange.upper + ) + onVideoTargetBitrateChange(newVideoBitrate) + } + } + } + + /** + * Factory interface you must use to create a [SimpleBitrateRegulator] object. + * If you want to create a custom RTMP bitrate regulation implementation, create a factory that + * implements this interface. + */ + class Factory : IBitrateRegulator.Factory { + /** + * Creates a [SimpleBitrateRegulator] object from given parameters + * + * @param metricsTracker endpoint metrics tracker + * @param bitrateRegulatorConfig bitrate regulation configuration + * @param onVideoTargetBitrateChange call when you have to change video bitrate + * @param onAudioTargetBitrateChange call when you have to change audio bitrate + * @return a [SimpleBitrateRegulator] object + */ + override fun newBitrateRegulator( + metricsTracker: EndpointMetricsTracker, + bitrateRegulatorConfig: BitrateRegulatorConfig, + onVideoTargetBitrateChange: ((Int) -> Unit), + onAudioTargetBitrateChange: ((Int) -> Unit) + ): SimpleBitrateRegulator { + return SimpleBitrateRegulator( + metricsTracker, + bitrateRegulatorConfig, + onVideoTargetBitrateChange, + onAudioTargetBitrateChange + ) + } + } +} \ No newline at end of file diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorController.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorController.kt index 2ba195240..7f116dd4d 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorController.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorController.kt @@ -17,7 +17,7 @@ package io.github.thibaultbee.streampack.core.regulator.controllers import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig import io.github.thibaultbee.streampack.core.elements.encoders.IEncoder -import io.github.thibaultbee.streampack.core.elements.endpoints.IEndpoint +import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics import io.github.thibaultbee.streampack.core.pipelines.outputs.encoding.IEncodingPipelineOutput import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator import kotlinx.coroutines.CoroutineDispatcher @@ -27,16 +27,16 @@ import kotlinx.coroutines.CoroutineDispatcher * * @param audioEncoder the audio [IEncoder] * @param videoEncoder the video [IEncoder] - * @param endpoint the [IEndpoint] implementation + * @param metricsProvider the [WithEndpointMetrics] implementation * @param bitrateRegulatorFactory the [IBitrateRegulator.Factory] implementation. Use it to make your own bitrate regulator. * @param bitrateRegulatorConfig bitrate regulator configuration */ abstract class BitrateRegulatorController( - private val audioEncoder: IEncoder?, - private val videoEncoder: IEncoder?, - private val endpoint: IEndpoint, - private val bitrateRegulatorFactory: IBitrateRegulator.Factory, - private val bitrateRegulatorConfig: BitrateRegulatorConfig = BitrateRegulatorConfig() + protected val audioEncoder: IEncoder?, + protected val videoEncoder: IEncoder?, + protected val metricsProvider: WithEndpointMetrics<*>, + protected val bitrateRegulatorFactory: IBitrateRegulator.Factory, + protected val bitrateRegulatorConfig: BitrateRegulatorConfig = BitrateRegulatorConfig() ) : IBitrateRegulatorController { abstract class Factory : IBitrateRegulatorController.Factory { /** diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorControllers.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorControllers.kt new file mode 100644 index 000000000..7bfbacd1d --- /dev/null +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/BitrateRegulatorControllers.kt @@ -0,0 +1,38 @@ +/* + * 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.regulator.controllers + +import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator +import io.github.thibaultbee.streampack.core.regulator.SimpleBitrateRegulator +import io.github.thibaultbee.streampack.core.regulator.controllers.IntervalBitrateRegulatorController.Companion.DEFAULT_POLLING_TIME +import io.github.thibaultbee.streampack.core.regulator.controllers.IntervalBitrateRegulatorController.Factory +import kotlin.time.Duration + +/** + * A [IntervalBitrateRegulatorController.Factory] for [IBitrateRegulator]. + * + * @param bitrateRegulatorFactory the [IBitrateRegulator.Factory] implementation. Use it to make your own bitrate regulator. + * @param bitrateRegulatorConfig bitrate regulator configuration + * @param pollingTime delay between each call to [IBitrateRegulator.update] + */ +fun intervalBitrateRegulatorControllerFactory( + bitrateRegulatorFactory: IBitrateRegulator.Factory = SimpleBitrateRegulator.Factory(), + bitrateRegulatorConfig: BitrateRegulatorConfig = BitrateRegulatorConfig(), + pollingTime: Duration = DEFAULT_POLLING_TIME +) = Factory( + bitrateRegulatorFactory, bitrateRegulatorConfig, pollingTime +) diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/SimpleBitrateRegulatorController.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/IntervalBitrateRegulatorController.kt similarity index 76% rename from core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/SimpleBitrateRegulatorController.kt rename to core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/IntervalBitrateRegulatorController.kt index cb39e2d3d..6a3f99ca1 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/SimpleBitrateRegulatorController.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/regulator/controllers/IntervalBitrateRegulatorController.kt @@ -17,56 +17,61 @@ package io.github.thibaultbee.streampack.core.regulator.controllers import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig import io.github.thibaultbee.streampack.core.elements.encoders.IEncoder -import io.github.thibaultbee.streampack.core.elements.endpoints.IEndpoint +import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics +import io.github.thibaultbee.streampack.core.elements.metrics.createMetricsTracker import io.github.thibaultbee.streampack.core.elements.utils.CoroutineScheduler import io.github.thibaultbee.streampack.core.pipelines.outputs.encoding.IConfigurableAudioEncodingPipelineOutput import io.github.thibaultbee.streampack.core.pipelines.outputs.encoding.IConfigurableVideoEncodingPipelineOutput import io.github.thibaultbee.streampack.core.pipelines.outputs.encoding.IEncodingPipelineOutput import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator import kotlinx.coroutines.CoroutineDispatcher +import kotlin.time.Duration +import kotlin.time.Duration.Companion.milliseconds /** - * A [BitrateRegulatorController] implementation that triggers [IBitrateRegulator.update] every [pollingTimeInMs]. + * A [BitrateRegulatorController] implementation that triggers [IBitrateRegulator.update] every [pollingTime]. * * @param audioEncoder the audio [IEncoder] * @param videoEncoder the video [IEncoder] - * @param endpoint the [IEndpoint] implementation + * @param metricsProvider the [WithEndpointMetrics] implementation * @param bitrateRegulatorFactory the [IBitrateRegulator.Factory] implementation. Use it to make your own bitrate regulator. * @param bitrateRegulatorConfig bitrate regulator configuration - * @param pollingTimeInMs delay between each call to [IBitrateRegulator.update] + * @param pollingTime delay between each call to [IBitrateRegulator.update] */ -open class SimpleBitrateRegulatorController( +class IntervalBitrateRegulatorController( audioEncoder: IEncoder?, videoEncoder: IEncoder, - endpoint: IEndpoint, + metricsProvider: WithEndpointMetrics<*>, bitrateRegulatorFactory: IBitrateRegulator.Factory, coroutineDispatcher: CoroutineDispatcher, bitrateRegulatorConfig: BitrateRegulatorConfig = BitrateRegulatorConfig(), - pollingTimeInMs: Long = DEFAULT_POLLING_TIME_IN_MS + pollingTime: Duration = DEFAULT_POLLING_TIME ) : BitrateRegulatorController( audioEncoder, videoEncoder, - endpoint, + metricsProvider, bitrateRegulatorFactory, bitrateRegulatorConfig ) { + private val metricsTracker = metricsProvider.createMetricsTracker() + /** * Bitrate regulator. Calls regularly by [scheduler]. Don't call it otherwise or you might break regulation. */ private val bitrateRegulator = bitrateRegulatorFactory.newBitrateRegulator( + metricsTracker, bitrateRegulatorConfig, - { + onVideoTargetBitrateChange = { videoEncoder.bitrate = it }, - { /* Do nothing for audio */ } + onAudioTargetBitrateChange = { /* Do nothing for audio */ } ) /** * Scheduler for bitrate regulation */ - private val scheduler = CoroutineScheduler(pollingTimeInMs, coroutineDispatcher) { + private val scheduler = CoroutineScheduler(pollingTime, coroutineDispatcher) { bitrateRegulator.update( - endpoint.metrics, videoEncoder.bitrate, audioEncoder?.bitrate ?: 0 ) @@ -81,13 +86,13 @@ open class SimpleBitrateRegulatorController( } companion object { - const val DEFAULT_POLLING_TIME_IN_MS = 500L + val DEFAULT_POLLING_TIME = 500.milliseconds } class Factory( private val bitrateRegulatorFactory: IBitrateRegulator.Factory, private val bitrateRegulatorConfig: BitrateRegulatorConfig = BitrateRegulatorConfig(), - private val pollingTimeInMs: Long = DEFAULT_POLLING_TIME_IN_MS + private val pollingTime: Duration = DEFAULT_POLLING_TIME ) : BitrateRegulatorController.Factory() { override fun newBitrateRegulatorController( pipelineOutput: IEncodingPipelineOutput, @@ -106,14 +111,15 @@ open class SimpleBitrateRegulatorController( } else { null } - return SimpleBitrateRegulatorController( + val endpoint = pipelineOutput.endpoint as WithEndpointMetrics<*> + return IntervalBitrateRegulatorController( audioEncoder, videoEncoder, - pipelineOutput.endpoint, + endpoint, bitrateRegulatorFactory, coroutineDispatcher, bitrateRegulatorConfig, - pollingTimeInMs + pollingTime ) } } diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/ISingleStreamer.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/ISingleStreamer.kt index f0d14fe8c..830680e2c 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/ISingleStreamer.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/ISingleStreamer.kt @@ -102,11 +102,5 @@ interface IVideoSingleStreamer : IVideoStreamer, ISingleStreamer { /** * Adds a bitrate regulator controller to the streamer. */ - fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) - - /** - * Removes the bitrate regulator controller from the streamer. - */ - fun removeBitrateRegulatorController() + var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory? } - diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamer.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamer.kt index cfd1b71f1..84bc5a475 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamer.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamer.kt @@ -269,10 +269,11 @@ class SingleStreamer( override suspend fun release() = streamer.release() - override fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) = - streamer.addBitrateRegulatorController(controllerFactory) - - override fun removeBitrateRegulatorController() = streamer.removeBitrateRegulatorController() + override var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory? + get() = streamer.bitrateRegulatorControllerFactory + set(value) { + streamer.bitrateRegulatorControllerFactory = value + } } diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamerImpl.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamerImpl.kt index 62af496b8..cb5a5f425 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamerImpl.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/SingleStreamerImpl.kt @@ -262,17 +262,12 @@ internal class SingleStreamerImpl( /** * Adds a bitrate regulator controller. - * - * Limitation: it is only available for SRT for now. */ - override fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) = - pipelineOutput.addBitrateRegulatorController(controllerFactory) - - /** - * Removes the bitrate regulator controller. - */ - override fun removeBitrateRegulatorController() = - pipelineOutput.removeBitrateRegulatorController() + override var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory? + get() = pipelineOutput.bitrateRegulatorControllerFactory + set(value) { + pipelineOutput.bitrateRegulatorControllerFactory = value + } companion object Companion { const val TAG = "SingleStreamer" diff --git a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/VideoOnlySingleStreamer.kt b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/VideoOnlySingleStreamer.kt index c224ba146..298c40b35 100644 --- a/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/VideoOnlySingleStreamer.kt +++ b/core/src/main/java/io/github/thibaultbee/streampack/core/streamers/single/VideoOnlySingleStreamer.kt @@ -193,8 +193,9 @@ class VideoOnlySingleStreamer( override suspend fun release() = streamer.release() - override fun addBitrateRegulatorController(controllerFactory: IBitrateRegulatorController.Factory) = - streamer.addBitrateRegulatorController(controllerFactory) - - override fun removeBitrateRegulatorController() = streamer.removeBitrateRegulatorController() + override var bitrateRegulatorControllerFactory: IBitrateRegulatorController.Factory? + get() = streamer.bitrateRegulatorControllerFactory + set(value) { + streamer.bitrateRegulatorControllerFactory = value + } } \ No newline at end of file diff --git a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/data/storage/DataStoreRepository.kt b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/data/storage/DataStoreRepository.kt index ddab33b3d..a170631d3 100644 --- a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/data/storage/DataStoreRepository.kt +++ b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/data/storage/DataStoreRepository.kt @@ -164,7 +164,7 @@ class DataStoreRepository( EndpointType.RTMP -> { val url = preferences[stringPreferencesKey(context.getString(R.string.rtmp_server_url_key))] - ?: context.getString(R.string.default_rtmp_url) + ?: context.getString(R.string.rtmp_default_url) UriMediaDescriptor(context, url) } } @@ -172,19 +172,71 @@ class DataStoreRepository( val bitrateRegulatorConfigFlow: Flow = dataStore.data.map { preferences -> + val endpointTypeId = + preferences[stringPreferencesKey(context.getString(R.string.endpoint_type_key))]?.toInt() + ?: EndpointType.SRT.id + val isBitrateRegulatorEnable = - preferences[booleanPreferencesKey(context.getString(R.string.srt_server_enable_bitrate_regulation_key))] + preferences[booleanPreferencesKey( + context.getString( + when (endpointTypeId) { + EndpointType.SRT.id -> { + R.string.srt_server_enable_bitrate_regulation_key + } + + EndpointType.RTMP.id -> { + R.string.rtmp_server_enable_bitrate_regulation_key + } + + else -> { + throw IllegalArgumentException("Unknown endpoint type") + } + } + ) + )] ?: true if (!isBitrateRegulatorEnable) { return@map null } val videoMinBitrate = - preferences[intPreferencesKey(context.getString(R.string.srt_server_video_min_bitrate_key))]?.toInt() + preferences[intPreferencesKey( + context.getString( + when (endpointTypeId) { + EndpointType.SRT.id -> { + R.string.srt_server_video_min_bitrate_key + } + + EndpointType.RTMP.id -> { + R.string.rtmp_server_video_min_bitrate_key + } + + else -> { + throw IllegalArgumentException("Unknown endpoint type") + } + } + ) + )] ?.times(1000) ?: 300000 val videoMaxBitrate = - preferences[intPreferencesKey(context.getString(R.string.srt_server_video_target_bitrate_key))]?.toInt() + preferences[intPreferencesKey( + context.getString( + when (endpointTypeId) { + EndpointType.SRT.id -> { + R.string.srt_server_video_target_bitrate_key + } + + EndpointType.RTMP.id -> { + R.string.rtmp_server_video_target_bitrate_key + } + + else -> { + throw IllegalArgumentException("Unknown endpoint type") + } + } + ) + )] ?.times(1000) ?: 10000000 BitrateRegulatorConfig( diff --git a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/main/PreviewViewModel.kt b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/main/PreviewViewModel.kt index 0b3dbd592..70d4b40b2 100644 --- a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/main/PreviewViewModel.kt +++ b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/main/PreviewViewModel.kt @@ -44,6 +44,10 @@ import io.github.thibaultbee.streampack.app.utils.setNextCameraId import io.github.thibaultbee.streampack.app.utils.toggleBackToFront import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.UriMediaDescriptor import io.github.thibaultbee.streampack.core.elements.endpoints.MediaSinkType +import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics +import io.github.thibaultbee.streampack.core.elements.metrics.metricsFlow +import io.github.thibaultbee.streampack.app.utils.formatBitrate +import io.github.thibaultbee.streampack.core.elements.metrics.writtenBitrateInBps import io.github.thibaultbee.streampack.core.elements.sources.audio.audiorecord.IAudioRecordSource import io.github.thibaultbee.streampack.core.elements.sources.audio.audiorecord.MicrophoneSourceFactory import io.github.thibaultbee.streampack.core.elements.sources.video.bitmap.BitmapSourceFactory @@ -58,6 +62,7 @@ import io.github.thibaultbee.streampack.core.interfaces.IWithVideoSource import io.github.thibaultbee.streampack.core.interfaces.releaseBlocking import io.github.thibaultbee.streampack.core.interfaces.startStream import io.github.thibaultbee.streampack.core.pipelines.StreamerPipeline +import io.github.thibaultbee.streampack.core.regulator.controllers.intervalBitrateRegulatorControllerFactory import io.github.thibaultbee.streampack.core.streamers.single.IAudioSingleStreamer import io.github.thibaultbee.streampack.core.streamers.single.IVideoSingleStreamer import io.github.thibaultbee.streampack.core.streamers.single.SingleStreamer @@ -65,23 +70,27 @@ import io.github.thibaultbee.streampack.core.streamers.single.VideoOnlySingleStr import io.github.thibaultbee.streampack.core.streamers.single.withAudio import io.github.thibaultbee.streampack.core.streamers.single.withVideo import io.github.thibaultbee.streampack.core.utils.extensions.isClosedException -import io.github.thibaultbee.streampack.ext.srt.regulator.controllers.simpleSrtBitrateRegulatorControllerFactory +import io.github.thibaultbee.streampack.ext.srt.regulator.controllers.intervalSrtBitrateRegulatorControllerFactory import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.drop +import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.map import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +@OptIn(ExperimentalCoroutinesApi::class) class PreviewViewModel(private val application: Application) : ObservableViewModel() { private val storageRepository = DataStoreRepository(application, application.dataStore) private val rotationRepository = RotationRepository.getInstance(application) @@ -91,7 +100,7 @@ class PreviewViewModel(private val application: Application) : ObservableViewMod private val buildStreamerUseCase = BuildStreamerUseCase(application) private val streamerFlow = - MutableStateFlow( + MutableStateFlow( buildFirstStreamer(runBlocking { storageRepository.isAudioEnableFlow.first() }) ) @@ -150,6 +159,12 @@ class PreviewViewModel(private val application: Application) : ObservableViewMod private val _isTryingConnectionLiveData = MutableLiveData() val isTryingConnectionLiveData: LiveData = _isTryingConnectionLiveData + private val _instantBitrateLiveData = MutableLiveData("0 bps") + val instantBitrateLiveData: LiveData = _instantBitrateLiveData + + private val _packetLossLiveData = MutableLiveData("0 / 0 (0%)") + val packetLossLiveData: LiveData = _packetLossLiveData + private val videoSourceMutex = Mutex() private var startStreamJob: Job? = null @@ -262,6 +277,26 @@ class PreviewViewModel(private val application: Application) : ObservableViewMod } } } + viewModelScope.launch { + streamerFlow.flatMapLatest { streamer -> + val endpointMetrics = streamer.endpoint as? WithEndpointMetrics<*> + endpointMetrics?.metricsFlow() ?: emptyFlow() + }.collect { metrics -> + val formattedBitrate = metrics.instant.writtenBitrateInBps.formatBitrate() + _instantBitrateLiveData.postValue(formattedBitrate) + val lost = + metrics.cumulative.packetsWriteDropped + metrics.cumulative.packetsWriteLost + val total = metrics.cumulative.packetsWritten + lost + val packetLossStr = if (total == 0L) { + "0 / 0 (0%)" + } else { + val percentage = (lost * 100f / total) + val formattedPercentage = String.format(java.util.Locale.US, "%.2f", percentage) + "$lost / $total ($formattedPercentage%)" + } + _packetLossLiveData.postValue(packetLossStr) + } + } } private fun buildFirstStreamer(isAudioEnable: Boolean): IVideoSingleStreamer { @@ -312,16 +347,32 @@ class PreviewViewModel(private val application: Application) : ObservableViewMod val descriptor = storageRepository.endpointDescriptorFlow.first() streamer.startStream(descriptor) - if (descriptor.type.sinkType == MediaSinkType.SRT) { + if ((descriptor.type.sinkType == MediaSinkType.RTMP) || (descriptor.type.sinkType == MediaSinkType.SRT)) { val bitrateRegulatorConfig = storageRepository.bitrateRegulatorConfigFlow.first() if (bitrateRegulatorConfig != null) { Log.i(TAG, "Add bitrate regulator controller") - streamer.addBitrateRegulatorController( - simpleSrtBitrateRegulatorControllerFactory( - bitrateRegulatorConfig = bitrateRegulatorConfig - ) - ) + val controllerFactory = + when (descriptor.type.sinkType) { + MediaSinkType.RTMP -> { + intervalBitrateRegulatorControllerFactory( + bitrateRegulatorConfig = bitrateRegulatorConfig + ) + } + + MediaSinkType.SRT -> { + intervalSrtBitrateRegulatorControllerFactory( + bitrateRegulatorConfig = bitrateRegulatorConfig + ) + } + + else -> { + null + } + } + controllerFactory?.let { + streamer.bitrateRegulatorControllerFactory = it + } ?: Log.e(TAG, "Controller factory is null") } } } catch (e: CancellationException) { @@ -344,7 +395,6 @@ class PreviewViewModel(private val application: Application) : ObservableViewMod try { streamer.stopStream() streamer.close() - streamer.removeBitrateRegulatorController() } catch (e: Throwable) { Log.e(TAG, "stopStream failed", e) } diff --git a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/settings/SettingsFragment.kt b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/settings/SettingsFragment.kt index 2a8079eda..04a3b1b1b 100644 --- a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/settings/SettingsFragment.kt +++ b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/ui/settings/SettingsFragment.kt @@ -131,18 +131,30 @@ class SettingsFragment : PreferenceFragmentCompat() { this.findPreference(getString(R.string.srt_server_port_key))!! } - private val serverEnableBitrateRegulationPreference: SwitchPreference by lazy { + private val srtServerEnableBitrateRegulationPreference: SwitchPreference by lazy { this.findPreference(getString(R.string.srt_server_enable_bitrate_regulation_key))!! } - private val serverTargetVideoBitratePreference: SeekBarPreference by lazy { + private val srtServerTargetVideoBitratePreference: SeekBarPreference by lazy { this.findPreference(getString(R.string.srt_server_video_target_bitrate_key))!! } - private val serverMinVideoBitratePreference: SeekBarPreference by lazy { + private val srtServerMinVideoBitratePreference: SeekBarPreference by lazy { this.findPreference(getString(R.string.srt_server_video_min_bitrate_key))!! } + private val rtmpServerEnableBitrateRegulationPreference: SwitchPreference by lazy { + this.findPreference(getString(R.string.rtmp_server_enable_bitrate_regulation_key))!! + } + + private val rtmpServerTargetVideoBitratePreference: SeekBarPreference by lazy { + this.findPreference(getString(R.string.rtmp_server_video_target_bitrate_key))!! + } + + private val rtmpServerMinVideoBitratePreference: SeekBarPreference by lazy { + this.findPreference(getString(R.string.rtmp_server_video_min_bitrate_key))!! + } + private val fileNamePreference: EditTextPreference by lazy { this.findPreference(getString(R.string.file_name_key))!! } @@ -417,26 +429,50 @@ class SettingsFragment : PreferenceFragmentCompat() { editText.filters = arrayOf(InputFilter.LengthFilter(5)) } - serverTargetVideoBitratePreference.isVisible = - serverEnableBitrateRegulationPreference.isChecked - serverMinVideoBitratePreference.isVisible = - serverEnableBitrateRegulationPreference.isChecked - serverEnableBitrateRegulationPreference.setOnPreferenceChangeListener { _, newValue -> - serverTargetVideoBitratePreference.isVisible = newValue as Boolean - serverMinVideoBitratePreference.isVisible = newValue + srtServerTargetVideoBitratePreference.isVisible = + srtServerEnableBitrateRegulationPreference.isChecked + srtServerMinVideoBitratePreference.isVisible = + srtServerEnableBitrateRegulationPreference.isChecked + srtServerEnableBitrateRegulationPreference.setOnPreferenceChangeListener { _, newValue -> + srtServerTargetVideoBitratePreference.isVisible = newValue as Boolean + srtServerMinVideoBitratePreference.isVisible = newValue + true + } + + srtServerTargetVideoBitratePreference.setOnPreferenceChangeListener { _, newValue -> + if ((newValue as Int) < srtServerMinVideoBitratePreference.value) { + srtServerMinVideoBitratePreference.value = newValue + } + true + } + + srtServerMinVideoBitratePreference.setOnPreferenceChangeListener { _, newValue -> + if ((newValue as Int) > srtServerTargetVideoBitratePreference.value) { + srtServerTargetVideoBitratePreference.value = newValue + } + true + } + + rtmpServerTargetVideoBitratePreference.isVisible = + rtmpServerEnableBitrateRegulationPreference.isChecked + rtmpServerMinVideoBitratePreference.isVisible = + rtmpServerEnableBitrateRegulationPreference.isChecked + rtmpServerEnableBitrateRegulationPreference.setOnPreferenceChangeListener { _, newValue -> + rtmpServerTargetVideoBitratePreference.isVisible = newValue as Boolean + rtmpServerMinVideoBitratePreference.isVisible = newValue true } - serverTargetVideoBitratePreference.setOnPreferenceChangeListener { _, newValue -> - if ((newValue as Int) < serverMinVideoBitratePreference.value) { - serverMinVideoBitratePreference.value = newValue + rtmpServerTargetVideoBitratePreference.setOnPreferenceChangeListener { _, newValue -> + if ((newValue as Int) < rtmpServerMinVideoBitratePreference.value) { + rtmpServerMinVideoBitratePreference.value = newValue } true } - serverMinVideoBitratePreference.setOnPreferenceChangeListener { _, newValue -> - if ((newValue as Int) > serverTargetVideoBitratePreference.value) { - serverTargetVideoBitratePreference.value = newValue + rtmpServerMinVideoBitratePreference.setOnPreferenceChangeListener { _, newValue -> + if ((newValue as Int) > rtmpServerTargetVideoBitratePreference.value) { + rtmpServerTargetVideoBitratePreference.value = newValue } true } @@ -459,7 +495,7 @@ class SettingsFragment : PreferenceFragmentCompat() { // Update file extension if (endpoint.hasFileCapabilities) { // Remove previous extension - FileExtension.entries.forEach { + FileExtension.entries.forEach { _ -> fileNamePreference.text = fileNamePreference.text?.substringBeforeLast(".") } // Add correct extension diff --git a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/utils/Extensions.kt b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/utils/Extensions.kt index a712b254e..c565ac65d 100644 --- a/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/utils/Extensions.kt +++ b/demos/camera/src/main/java/io/github/thibaultbee/streampack/app/utils/Extensions.kt @@ -98,3 +98,11 @@ fun String.appendIfNotEndsWith(suffix: String): String { val Range<*>.isEmpty: Boolean get() = upper == lower + +fun Long.formatBitrate(): String { + return when { + this >= 1_000_000 -> String.format(java.util.Locale.US, "%.2f Mbps", this / 1_000_000.0) + this >= 1_000 -> String.format(java.util.Locale.US, "%.2f kbps", this / 1_000.0) + else -> "$this bps" + } +} diff --git a/demos/camera/src/main/res/layout/main_fragment.xml b/demos/camera/src/main/res/layout/main_fragment.xml index 03f03eef6..36a8a4bb5 100644 --- a/demos/camera/src/main/res/layout/main_fragment.xml +++ b/demos/camera/src/main/res/layout/main_fragment.xml @@ -1,7 +1,8 @@ + xmlns:app="http://schemas.android.com/apk/res-auto" + xmlns:tools="http://schemas.android.com/tools"> @@ -27,6 +28,31 @@ app:position="center" app:scaleMode="fill" /> + + + + + + + rtmp_server_key RTMP Server rtmp_server_url_key - rtmp://192.168.1.192/s/streamKey + rtmp://192.168.1.192/s/streamKey + rtmp_server_enable_bitrate_regulation_key + Enable bitrate regulation + rtmp_server_video_target_bitrate_key + Video target bitrate (kb/s) + rtmp_server_video_min_bitrate_key + Video minimum bitrate (kb/s) + URL file_endpoint_key diff --git a/demos/camera/src/main/res/xml/root_preferences.xml b/demos/camera/src/main/res/xml/root_preferences.xml index e47de9ad3..58a3a89b7 100644 --- a/demos/camera/src/main/res/xml/root_preferences.xml +++ b/demos/camera/src/main/res/xml/root_preferences.xml @@ -167,11 +167,33 @@ app:title="@string/rtmp_server"> + + + + + = _isOpenFlow.asStateFlow() diff --git a/extensions/rtmp/build.gradle.kts b/extensions/rtmp/build.gradle.kts index beca775cf..1be6ad482 100644 --- a/extensions/rtmp/build.gradle.kts +++ b/extensions/rtmp/build.gradle.kts @@ -14,7 +14,7 @@ dependencies { implementation(project(":streampack-core")) implementation(project(":streampack-flv")) - implementation(libs.komedia.komuxer.rtmp) + api(libs.komedia.komuxer.rtmp) implementation(libs.kotlinx.coroutines.android) implementation(libs.androidx.core.ktx) diff --git a/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpoint.kt b/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpoint.kt index 1e2b581ea..85076aa91 100644 --- a/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpoint.kt +++ b/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpoint.kt @@ -16,12 +16,16 @@ package io.github.thibaultbee.streampack.ext.rtmp.elements.endpoints import android.content.Context +import io.github.komedia.komuxer.amf.AmfVersion import io.github.komedia.komuxer.flv.tags.FLVTag +import io.github.komedia.komuxer.flv.tags.audio.AudioData +import io.github.komedia.komuxer.flv.tags.video.VideoData import io.github.komedia.komuxer.rtmp.RtmpConnectionBuilder import io.github.komedia.komuxer.rtmp.client.RtmpClient import io.github.komedia.komuxer.rtmp.client.RtmpClientSettings import io.github.komedia.komuxer.rtmp.connect import io.github.komedia.komuxer.rtmp.messages.command.StreamPublishType +import io.github.komedia.komuxer.rtmp.util.metrics.RtmpMetrics 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.encoders.CodecConfig @@ -29,6 +33,7 @@ import io.github.thibaultbee.streampack.core.elements.endpoints.ClosedException import io.github.thibaultbee.streampack.core.elements.endpoints.IEndpoint import io.github.thibaultbee.streampack.core.elements.endpoints.IEndpointInternal import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpoint.EndpointInfo +import io.github.thibaultbee.streampack.core.elements.metrics.WithEndpointMetrics import io.github.thibaultbee.streampack.core.elements.utils.ChannelWithCloseableData import io.github.thibaultbee.streampack.core.elements.utils.useConsumeEach import io.github.thibaultbee.streampack.core.logger.Logger @@ -53,18 +58,43 @@ import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import kotlinx.io.EOFException import java.io.IOException +import kotlin.concurrent.atomics.ExperimentalAtomicApi /** * An endpoint that send frame to an RTMP server. */ +@OptIn(ExperimentalAtomicApi::class) class RtmpEndpoint internal constructor( defaultDispatcher: CoroutineDispatcher, val ioDispatcher: CoroutineDispatcher -) : IEndpointInternal { +) : IEndpointInternal, WithEndpointMetrics { private val coroutineScope = CoroutineScope(SupervisorJob() + defaultDispatcher) private val mutex = Mutex() + private val metricsLock = Any() + private var frameDropped = 0L + private var audioFrameDropped = 0L + private var videoFrameDropped = 0L + + private var payloadSendDroppedSize = 0L + private var audioPayloadSendDroppedSize = 0L + private var videoPayloadSendDroppedSize = 0L + private val flvTagChannel = ChannelWithCloseableData( - 10 /* Arbitrary buffer size. TODO: add a parameter to set it */, BufferOverflow.DROP_OLDEST + 10 /* Arbitrary buffer size. TODO: add a parameter to set it */, BufferOverflow.DROP_OLDEST, + onUndeliveredElement = { flvTag -> + synchronized(metricsLock) { + val payloadSize = flvTag.data.getSize(AmfVersion.AMF0) + frameDropped++ + payloadSendDroppedSize += payloadSize + if (flvTag.data is AudioData) { + audioFrameDropped++ + audioPayloadSendDroppedSize += payloadSize + } else if (flvTag.data is VideoData) { + videoFrameDropped++ + videoPayloadSendDroppedSize += payloadSize + } + } + } ) private val flvTagBuilder = FlvTagBuilder(flvTagChannel) @@ -72,12 +102,27 @@ class RtmpEndpoint internal constructor( private val connectionBuilder = RtmpConnectionBuilder(selectorManager) private var rtmpClient: RtmpClient? = null - private var startUpTimestamp = INVALID_TIMESTAMP private val timestampMutex = Mutex() - override val metrics: Any - get() = TODO("Not yet implemented") + private val syncMetrics: RtmpMetrics + get() { + val metrics = + rtmpClient?.metrics ?: return RtmpMetrics.ZERO + return synchronized(metricsLock) { + metrics.copy( + messagesSendDropped = metrics.messagesSendDropped + frameDropped, + audioMessagesSendDropped = metrics.audioMessagesSendDropped + audioFrameDropped, + videoMessagesSendDropped = metrics.videoMessagesSendDropped + videoFrameDropped, + payloadSendDroppedSize = metrics.payloadSendDroppedSize + payloadSendDroppedSize, + audioPayloadSendDroppedSize = metrics.audioPayloadSendDroppedSize + audioPayloadSendDroppedSize, + videoPayloadSendDroppedSize = metrics.videoPayloadSendDroppedSize + videoPayloadSendDroppedSize + ) + } + } + + override val metrics: RtmpEndpointMetrics + get() = RtmpEndpointMetrics { syncMetrics } private val _isOpenFlow = MutableStateFlow(false) override val isOpenFlow = _isOpenFlow.asStateFlow() @@ -97,6 +142,18 @@ class RtmpEndpoint internal constructor( } } + private fun resetMetrics() { + synchronized(metricsLock) { + frameDropped = 0L + audioFrameDropped = 0L + videoFrameDropped = 0L + + payloadSendDroppedSize = 0L + audioPayloadSendDroppedSize = 0L + videoPayloadSendDroppedSize = 0L + } + } + private suspend fun safeClient(block: suspend (RtmpClient) -> T): T { val rtmpClient = requireNotNull(rtmpClient) { "Not opened" } require(!rtmpClient.isClosed) { "Connection closed" } @@ -114,6 +171,7 @@ class RtmpEndpoint internal constructor( return@withContext } + resetMetrics() val client = if (descriptor.connectInfo != null) { connectionBuilder.connect( descriptor.uri.toString(), diff --git a/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpointMetrics.kt b/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpointMetrics.kt new file mode 100644 index 000000000..93609a04b --- /dev/null +++ b/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/elements/endpoints/RtmpEndpointMetrics.kt @@ -0,0 +1,74 @@ +/* + * 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.ext.rtmp.elements.endpoints + +import io.github.komedia.komuxer.rtmp.util.metrics.RtmpMetrics +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetrics +import kotlin.time.Duration + +/** + * Creates a [RtmpEndpointMetrics] from a [metricsProvider]. + */ +fun RtmpEndpointMetrics(metricsProvider: () -> RtmpMetrics?): RtmpEndpointMetrics { + return RtmpEndpointMetrics(RtmpRawMetrics(metricsProvider)) +} + +/** + * Creates a [RtmpEndpointMetrics] from a [RtmpMetrics]. + */ +fun RtmpEndpointMetrics(rawMetrics: RtmpRawMetrics): RtmpEndpointMetrics { + val metrics = rawMetrics.rtmpMetrics + return RtmpEndpointMetrics( + uptime = metrics.uptime, + packetsWritten = metrics.messagesSent, + packetsWriteDropped = metrics.messagesSendDropped, + packetsWriteLost = 0L, + bytesWritten = metrics.totalBytesSent, + bytesWriteDropped = metrics.payloadSendDroppedSize, + rawMetrics = rawMetrics + ) +} + +/** + * Specific [EndpointMetrics] for RTMP protocol, based on [RtmpMetrics]. + */ +data class RtmpEndpointMetrics( + override val uptime: Duration, + override val packetsWritten: Long, + override val packetsWriteDropped: Long, + override val packetsWriteLost: Long, + override val bytesWritten: Long, + override val bytesWriteDropped: Long, + override val rawMetrics: RtmpRawMetrics +) : EndpointMetrics + + +/** + * Provides an access to internal RTMP metrics APIs. + */ +class RtmpRawMetrics internal constructor(private val metricsProvider: () -> RtmpMetrics?) { + /** + * Returns the [RtmpMetrics] if the client is available, otherwise null. + */ + val rtmpMetricsOrNull: RtmpMetrics? + get() = metricsProvider() +} + +/** + * Returns the [RtmpMetrics] if the client is available, otherwise [RtmpMetrics.ZERO]. + */ +val RtmpRawMetrics.rtmpMetrics: RtmpMetrics + get() = rtmpMetricsOrNull ?: RtmpMetrics.ZERO \ No newline at end of file diff --git a/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/regulator/RtmpBitrateRegulator.kt b/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/regulator/RtmpBitrateRegulator.kt new file mode 100644 index 000000000..c7acf37b4 --- /dev/null +++ b/extensions/rtmp/src/main/java/io/github/thibaultbee/streampack/ext/rtmp/regulator/RtmpBitrateRegulator.kt @@ -0,0 +1,68 @@ +/* + * 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.ext.rtmp.regulator + +import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetricsTracker +import io.github.thibaultbee.streampack.core.regulator.BitrateRegulator +import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator + +/** + * Base class of RTMP bitrate regulation implementation. + * + * If you want to implement your custom bitrate regulator, it must inherit from this class. + * The bitrate regulator object is created by streamers with the [IBitrateRegulator.Factory]. + * + * @param bitrateRegulatorConfig bitrate regulation configuration + * @param onVideoTargetBitrateChange call when you have to change video bitrate + * @param onAudioTargetBitrateChange call when you have to change audio bitrate + */ +abstract class RtmpBitrateRegulator( + metricsTracker: EndpointMetricsTracker, + bitrateRegulatorConfig: BitrateRegulatorConfig, + onVideoTargetBitrateChange: ((Int) -> Unit), + onAudioTargetBitrateChange: ((Int) -> Unit) +) : BitrateRegulator( + metricsTracker, + bitrateRegulatorConfig, + onVideoTargetBitrateChange, + onAudioTargetBitrateChange +) { + + + /** + * Factory interface you must use to create a [RtmpBitrateRegulator] object. + * If you want to create a custom RTMP bitrate regulation implementation, create a factory that + * implements this interface. + */ + interface Factory : IBitrateRegulator.Factory { + /** + * Creates a [RtmpBitrateRegulator] object from given parameters + * + * @param metricsTracker endpoint metrics tracker + * @param bitrateRegulatorConfig bitrate regulation configuration + * @param onVideoTargetBitrateChange call when you have to change video bitrate + * @param onAudioTargetBitrateChange call when you have to change audio bitrate + * @return a [RtmpBitrateRegulator] object + */ + override fun newBitrateRegulator( + metricsTracker: EndpointMetricsTracker, + bitrateRegulatorConfig: BitrateRegulatorConfig, + onVideoTargetBitrateChange: ((Int) -> Unit), + onAudioTargetBitrateChange: ((Int) -> Unit) + ): RtmpBitrateRegulator + } +} \ No newline at end of file diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointFactory.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointFactory.kt index 920181103..f91dbc8f6 100644 --- a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointFactory.kt +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointFactory.kt @@ -1,7 +1,23 @@ +/* + * 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.ext.srt.elements.endpoints import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.createDefaultTsServiceInfo import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpointFactory +import io.github.thibaultbee.streampack.core.elements.endpoints.composites.CompositeEndpointWithMetricsFactory 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 io.github.thibaultbee.streampack.ext.srt.elements.endpoints.composites.sinks.SrtSink @@ -19,7 +35,7 @@ fun SrtEndpointFactory( serviceInfo: TSServiceInfo = createDefaultTsServiceInfo(), coroutineDispatcher: CoroutineDispatcher ) = - CompositeEndpointFactory( + CompositeEndpointWithMetricsFactory( TsMuxer().apply { addService(serviceInfo) }, SrtSink(coroutineDispatcher) ) \ No newline at end of file diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointMetrics.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointMetrics.kt new file mode 100644 index 000000000..9c516290d --- /dev/null +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/SrtEndpointMetrics.kt @@ -0,0 +1,259 @@ +/* + * 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.ext.srt.elements.endpoints + +import io.github.thibaultbee.srtdroid.core.models.Stats +import io.github.thibaultbee.srtdroid.ktx.CoroutineSrtSocket +import io.github.thibaultbee.streampack.core.elements.metrics.BasicEndpointMetrics +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetrics +import io.github.thibaultbee.streampack.ext.srt.utils.SrtStatsHelper +import kotlin.time.Duration +import kotlin.time.Duration.Companion.milliseconds + +/** + * Creates a [SrtEndpointMetrics] from a [SrtRawMetrics]. + */ +fun SrtEndpointMetrics(rawMetrics: SrtRawMetrics): SrtEndpointMetrics { + val stats = rawMetrics.bstats(clear = false) + return SrtEndpointMetrics( + uptime = stats.msTimeStamp.milliseconds, + packetsWritten = stats.pktSentTotal, + packetsRead = stats.pktRecvTotal, + packetsWriteLost = stats.pktSndLossTotal.toLong(), + packetsReadLost = stats.pktRcvLossTotal, + packetsRetransmitted = stats.pktRetransTotal, + packetsWriteACK = stats.pktSentACKTotal, + packetsReadACK = stats.pktRecvACKTotal, + packetsWriteNAK = stats.pktSentNAKTotal, + packetsReadNAK = stats.pktRecvNAKTotal, + usWriteDuration = stats.usSndDurationTotal, + packetsWriteDropped = stats.pktSndDropTotal.toLong(), + packetsReadDropped = stats.pktRcvDropTotal, + packetsReadUndecrypt = stats.pktRcvUndecryptTotal, + bytesWritten = stats.byteSentTotal, + bytesRead = stats.byteRecvTotal, + bytesReadLost = stats.byteRcvLossTotal, + bytesRetransmitted = stats.byteRetransTotal, + bytesWriteDropped = stats.byteSndDropTotal, + bytesReadDropped = stats.byteRcvDropTotal, + bytesReadUndecrypt = stats.byteRcvUndecryptTotal, + rawMetrics = rawMetrics + ) +} + +/** + * Basic metrics for SRT protocol, without the raw metrics helper. + */ +interface SrtBasicEndpointMetrics : BasicEndpointMetrics { + /** + * The time since the entity is started + */ + override val uptime: Duration + /** + * The total number of written data packets, including retransmissions + */ + override val packetsWritten: Long + /** + * The total number of read packets + */ + val packetsRead: Long + /** + * The total number of lost packets (writer side) + */ + override val packetsWriteLost: Long + /** + * The total number of lost packets (reader side) + */ + val packetsReadLost: Int + /** + * The total number of retransmitted packets + */ + val packetsRetransmitted: Int + /** + * The total number of written ACK packets + */ + val packetsWriteACK: Int + /** + * The total number of read ACK packets + */ + val packetsReadACK: Int + /** + * The total number of written NAK packets + */ + val packetsWriteNAK: Int + /** + * The total number of read NAK packets + */ + val packetsReadNAK: Int + /** + * The total time duration when UDT is writing data (idle time exclusive) + */ + val usWriteDuration: Long + /** + * The number of too-late-to-write dropped packets + */ + override val packetsWriteDropped: Long + /** + * The number of too-late-to-play missing packets + */ + val packetsReadDropped: Int + /** + * The number of undecrypted packets + */ + val packetsReadUndecrypt: Int + /** + * The total number of written data bytes, including retransmissions + */ + override val bytesWritten: Long + /** + * The total number of read bytes + */ + val bytesRead: Long + /** + * The total number of lost bytes + */ + val bytesReadLost: Long + /** + * The total number of retransmitted bytes + */ + val bytesRetransmitted: Long + /** + * The number of too-late-to-write dropped bytes + */ + override val bytesWriteDropped: Long + /** + * The number of too-late-to-play missing bytes (estimate based on average packet size) + */ + val bytesReadDropped: Long + /** + * The number of undecrypted bytes + */ + val bytesReadUndecrypt: Long + + override operator fun minus(other: BasicEndpointMetrics): BasicEndpointMetrics { + if (other !is SrtBasicEndpointMetrics) return super.minus(other) + + return object : SrtBasicEndpointMetrics { + override val uptime = this@SrtBasicEndpointMetrics.uptime - other.uptime + override val packetsWritten = this@SrtBasicEndpointMetrics.packetsWritten - other.packetsWritten + override val packetsRead = this@SrtBasicEndpointMetrics.packetsRead - other.packetsRead + override val packetsWriteLost = this@SrtBasicEndpointMetrics.packetsWriteLost - other.packetsWriteLost + override val packetsReadLost = this@SrtBasicEndpointMetrics.packetsReadLost - other.packetsReadLost + override val packetsRetransmitted = this@SrtBasicEndpointMetrics.packetsRetransmitted - other.packetsRetransmitted + override val packetsWriteACK = this@SrtBasicEndpointMetrics.packetsWriteACK - other.packetsWriteACK + override val packetsReadACK = this@SrtBasicEndpointMetrics.packetsReadACK - other.packetsReadACK + override val packetsWriteNAK = this@SrtBasicEndpointMetrics.packetsWriteNAK - other.packetsWriteNAK + override val packetsReadNAK = this@SrtBasicEndpointMetrics.packetsReadNAK - other.packetsReadNAK + override val usWriteDuration = this@SrtBasicEndpointMetrics.usWriteDuration - other.usWriteDuration + override val packetsWriteDropped = this@SrtBasicEndpointMetrics.packetsWriteDropped - other.packetsWriteDropped + override val packetsReadDropped = this@SrtBasicEndpointMetrics.packetsReadDropped - other.packetsReadDropped + override val packetsReadUndecrypt = this@SrtBasicEndpointMetrics.packetsReadUndecrypt - other.packetsReadUndecrypt + override val bytesWritten = this@SrtBasicEndpointMetrics.bytesWritten - other.bytesWritten + override val bytesRead = this@SrtBasicEndpointMetrics.bytesRead - other.bytesRead + override val bytesReadLost = this@SrtBasicEndpointMetrics.bytesReadLost - other.bytesReadLost + override val bytesRetransmitted = this@SrtBasicEndpointMetrics.bytesRetransmitted - other.bytesRetransmitted + override val bytesWriteDropped = this@SrtBasicEndpointMetrics.bytesWriteDropped - other.bytesWriteDropped + override val bytesReadDropped = this@SrtBasicEndpointMetrics.bytesReadDropped - other.bytesReadDropped + override val bytesReadUndecrypt = this@SrtBasicEndpointMetrics.bytesReadUndecrypt - other.bytesReadUndecrypt + } + } +} + +/** + * Specific [EndpointMetrics] for SRT protocol, based on [SrtRawMetrics]. + */ +data class SrtEndpointMetrics( + override val uptime: Duration, + override val packetsWritten: Long, + override val packetsRead: Long, + override val packetsWriteLost: Long, + override val packetsReadLost: Int, + override val packetsRetransmitted: Int, + override val packetsWriteACK: Int, + override val packetsReadACK: Int, + override val packetsWriteNAK: Int, + override val packetsReadNAK: Int, + override val usWriteDuration: Long, + override val packetsWriteDropped: Long, + override val packetsReadDropped: Int, + override val packetsReadUndecrypt: Int, + override val bytesWritten: Long, + override val bytesRead: Long, + override val bytesReadLost: Long, + override val bytesRetransmitted: Long, + override val bytesWriteDropped: Long, + override val bytesReadDropped: Long, + override val bytesReadUndecrypt: Long, + /** + * Raw SRT socket helper + */ + override val rawMetrics: SrtRawMetrics, +) : SrtBasicEndpointMetrics, EndpointMetrics + +/** + * Provides an access to internal SRT APIs. + */ +class SrtRawMetrics internal constructor(private val socketProvider: () -> CoroutineSrtSocket?) { + /** + * Reports the current statistics. + * + * **See Also:** [srt_bstats](https://github.com/Haivision/srt/blob/master/docs/API/API-functions.md#srt_bstats) + * + * @param clear true if the statistics should be cleared after retrieval + * @return the current [Stats] or null if the socket is not connected + */ + fun bstatsOrNull(clear: Boolean): Stats? { + return socketProvider()?.bstats(clear) + } + + /** + * Reports the current statistics. + * + * **See Also:** [srt_bistats](https://github.com/Haivision/srt/blob/master/docs/API/API-functions.md#srt_bistats) + * + * @param clear true if the statistics should be cleared after retrieval + * @param instantaneous true if the statistics should use instant data, not moving averages + * @return the current [Stats] if the socket is not connected + */ + fun bistatsOrNull(clear: Boolean, instantaneous: Boolean): Stats? { + return socketProvider()?.bistats(clear, instantaneous) + } +} + +/** + * Reports the current statistics. + * + * **See Also:** [srt_bstats](https://github.com/Haivision/srt/blob/master/docs/API/API-functions.md#srt_bstats) + * + * @param clear true if the statistics should be cleared after retrieval + * @return the current [Stats] or [io.github.thibaultbee.streampack.ext.srt.utils.SrtStatsHelper.ZERO] if the socket is not connected + */ +fun SrtRawMetrics.bstats(clear: Boolean): Stats { + return bstatsOrNull(clear) ?: SrtStatsHelper.ZERO +} + +/** + * Reports the current statistics. + * + * **See Also:** [srt_bistats](https://github.com/Haivision/srt/blob/master/docs/API/API-functions.md#srt_bistats) + * + * @param clear true if the statistics should be cleared after retrieval + * @param instantaneous true if the statistics should use instant data, not moving averages + * @return the current [Stats] or [SrtStatsHelper.ZERO] if the socket is not connected + */ +fun SrtRawMetrics.bistats(clear: Boolean, instantaneous: Boolean): Stats { + return bistatsOrNull(clear, instantaneous) ?: SrtStatsHelper.ZERO +} \ No newline at end of file diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/composites/sinks/SrtSink.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/composites/sinks/SrtSink.kt index f5659ffee..1fdc7f5f5 100644 --- a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/composites/sinks/SrtSink.kt +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/elements/endpoints/composites/sinks/SrtSink.kt @@ -20,7 +20,6 @@ import io.github.thibaultbee.srtdroid.core.enums.SockOpt import io.github.thibaultbee.srtdroid.core.enums.Transtype import io.github.thibaultbee.srtdroid.core.models.MsgCtrl import io.github.thibaultbee.srtdroid.core.models.SrtUrl.Mode -import io.github.thibaultbee.srtdroid.core.models.Stats import io.github.thibaultbee.srtdroid.ktx.CoroutineSrtSocket import io.github.thibaultbee.srtdroid.ktx.connect import io.github.thibaultbee.streampack.core.configuration.mediadescriptor.MediaDescriptor @@ -30,13 +29,17 @@ import io.github.thibaultbee.streampack.core.elements.endpoints.composites.data. import io.github.thibaultbee.streampack.core.elements.endpoints.composites.data.SrtPacket import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.AbstractSink import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.SinkConfiguration +import io.github.thibaultbee.streampack.core.elements.endpoints.composites.sinks.ISinkWithMetricsInternal import io.github.thibaultbee.streampack.core.logger.Logger import io.github.thibaultbee.streampack.ext.srt.configuration.mediadescriptor.SrtMediaDescriptor +import io.github.thibaultbee.streampack.ext.srt.elements.endpoints.SrtRawMetrics +import io.github.thibaultbee.streampack.ext.srt.elements.endpoints.SrtEndpointMetrics import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.asStateFlow -class SrtSink(private val coroutineDispatcher: CoroutineDispatcher) : AbstractSink() { +class SrtSink(private val coroutineDispatcher: CoroutineDispatcher) : AbstractSink(), + ISinkWithMetricsInternal { override val supportedSinkTypes: List = listOf(MediaSinkType.SRT) private var socket: CoroutineSrtSocket? = null @@ -45,12 +48,13 @@ class SrtSink(private val coroutineDispatcher: CoroutineDispatcher) : AbstractSi private var bitrate = 0L + private val srtRawMetrics = SrtRawMetrics { socket } + /** * Get SRT stats */ - override val metrics: Stats - get() = socket?.bistats(clear = true, instantaneous = true) - ?: throw IllegalStateException("Socket is not initialized") + override val metrics: SrtEndpointMetrics + get() = SrtEndpointMetrics(srtRawMetrics) private val _isOpenFlow = MutableStateFlow(false) override val isOpenFlow = _isOpenFlow.asStateFlow() diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/DummySrtBitrateRegulator.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/DummySrtBitrateRegulator.kt index 65dfd6916..4e5cc3e2f 100644 --- a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/DummySrtBitrateRegulator.kt +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/DummySrtBitrateRegulator.kt @@ -15,8 +15,9 @@ */ package io.github.thibaultbee.streampack.ext.srt.regulator -import io.github.thibaultbee.srtdroid.core.models.Stats import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetricsTracker +import io.github.thibaultbee.streampack.ext.srt.elements.endpoints.SrtEndpointMetrics import kotlin.math.max import kotlin.math.min @@ -28,10 +29,12 @@ import kotlin.math.min * @param onAudioTargetBitrateChange not used in this implementation. */ class DummySrtBitrateRegulator( + metricsTracker: EndpointMetricsTracker, bitrateRegulatorConfig: BitrateRegulatorConfig, onVideoTargetBitrateChange: ((Int) -> Unit), onAudioTargetBitrateChange: ((Int) -> Unit) ) : SrtBitrateRegulator( + metricsTracker, bitrateRegulatorConfig, onVideoTargetBitrateChange, onAudioTargetBitrateChange @@ -42,7 +45,9 @@ class DummySrtBitrateRegulator( const val SEND_PACKET_THRESHOLD = 50 } - override fun update(stats: Stats, currentVideoBitrate: Int, currentAudioBitrate: Int) { + override fun update(currentVideoBitrate: Int, currentAudioBitrate: Int) { + val metrics = metricsTracker.cumulative as SrtEndpointMetrics + val stats = metrics.rawMetrics.bistatsOrNull(clear = true, instantaneous = true) ?: return val estimatedBandwidth = (stats.mbpsBandwidth * 1000000).toInt() if (currentVideoBitrate > bitrateRegulatorConfig.videoBitrateRange.lower) { @@ -72,18 +77,11 @@ class DummySrtBitrateRegulator( } if (newVideoBitrate != 0) { - onVideoTargetBitrateChange( - max( - newVideoBitrate, - bitrateRegulatorConfig.videoBitrateRange.lower - ) - ) // Don't go under videoBitrateRange.lower + onVideoTargetBitrateChange(newVideoBitrate) return } - } - - // Can bitrate go upper? - if (currentVideoBitrate < bitrateRegulatorConfig.videoBitrateRange.upper) { + // Can bitrate go upper? + } else if (currentVideoBitrate < bitrateRegulatorConfig.videoBitrateRange.upper) { val newVideoBitrate = when { (currentVideoBitrate + currentAudioBitrate) < estimatedBandwidth -> { currentVideoBitrate + min( @@ -96,12 +94,7 @@ class DummySrtBitrateRegulator( } if (newVideoBitrate != 0) { - onVideoTargetBitrateChange( - max( - newVideoBitrate, - bitrateRegulatorConfig.videoBitrateRange.lower - ) - ) // Don't go under videoBitrateRange.lower + onVideoTargetBitrateChange(newVideoBitrate) return } } @@ -115,17 +108,20 @@ class DummySrtBitrateRegulator( /** * Creates a [DummySrtBitrateRegulator] object from given parameters * + * @param metricsTracker endpoint metrics tracker * @param bitrateRegulatorConfig bitrate regulation configuration * @param onVideoTargetBitrateChange call when you have to change video bitrate * @param onAudioTargetBitrateChange call when you have to change audio bitrate * @return a [DummySrtBitrateRegulator] object */ override fun newBitrateRegulator( + metricsTracker: EndpointMetricsTracker, bitrateRegulatorConfig: BitrateRegulatorConfig, onVideoTargetBitrateChange: ((Int) -> Unit), onAudioTargetBitrateChange: ((Int) -> Unit) ): DummySrtBitrateRegulator { return DummySrtBitrateRegulator( + metricsTracker, bitrateRegulatorConfig, onVideoTargetBitrateChange, onAudioTargetBitrateChange diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/SrtBitrateRegulator.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/SrtBitrateRegulator.kt index dbfb53a84..79920fe30 100644 --- a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/SrtBitrateRegulator.kt +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/SrtBitrateRegulator.kt @@ -15,8 +15,8 @@ */ package io.github.thibaultbee.streampack.ext.srt.regulator -import io.github.thibaultbee.srtdroid.core.models.Stats import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig +import io.github.thibaultbee.streampack.core.elements.metrics.EndpointMetricsTracker import io.github.thibaultbee.streampack.core.regulator.BitrateRegulator import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator @@ -31,25 +31,17 @@ import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator * @param onAudioTargetBitrateChange call when you have to change audio bitrate */ abstract class SrtBitrateRegulator( + metricsTracker: EndpointMetricsTracker, bitrateRegulatorConfig: BitrateRegulatorConfig, onVideoTargetBitrateChange: ((Int) -> Unit), onAudioTargetBitrateChange: ((Int) -> Unit) ) : BitrateRegulator( + metricsTracker, bitrateRegulatorConfig, onVideoTargetBitrateChange, onAudioTargetBitrateChange ) { - override fun update(stats: Any, currentVideoBitrate: Int, currentAudioBitrate: Int) = - update(stats as Stats, currentVideoBitrate, currentAudioBitrate) - /** - * Call regularly to get new SRT stats - * - * @param stats SRT transmission stats - * @param currentVideoBitrate current video bitrate target in bits/s. - * @param currentAudioBitrate current audio bitrate target in bits/s. - */ - abstract fun update(stats: Stats, currentVideoBitrate: Int, currentAudioBitrate: Int) /** * Factory interface you must use to create a [SrtBitrateRegulator] object. @@ -60,12 +52,14 @@ abstract class SrtBitrateRegulator( /** * Creates a [SrtBitrateRegulator] object from given parameters * + * @param metricsTracker endpoint metrics tracker * @param bitrateRegulatorConfig bitrate regulation configuration * @param onVideoTargetBitrateChange call when you have to change video bitrate * @param onAudioTargetBitrateChange call when you have to change audio bitrate * @return a [SrtBitrateRegulator] object */ override fun newBitrateRegulator( + metricsTracker: EndpointMetricsTracker, bitrateRegulatorConfig: BitrateRegulatorConfig, onVideoTargetBitrateChange: ((Int) -> Unit), onAudioTargetBitrateChange: ((Int) -> Unit) diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/controllers/DefaultSrtBitrateRegulatorController.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/controllers/SrtBitrateRegulatorControllerFactories.kt similarity index 72% rename from extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/controllers/DefaultSrtBitrateRegulatorController.kt rename to extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/controllers/SrtBitrateRegulatorControllerFactories.kt index 5d56a545a..43fc3a2d5 100644 --- a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/controllers/DefaultSrtBitrateRegulatorController.kt +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/regulator/controllers/SrtBitrateRegulatorControllerFactories.kt @@ -16,27 +16,29 @@ package io.github.thibaultbee.streampack.ext.srt.regulator.controllers import io.github.thibaultbee.streampack.core.configuration.BitrateRegulatorConfig -import io.github.thibaultbee.streampack.core.regulator.controllers.SimpleBitrateRegulatorController -import io.github.thibaultbee.streampack.core.regulator.controllers.SimpleBitrateRegulatorController.Companion.DEFAULT_POLLING_TIME_IN_MS +import io.github.thibaultbee.streampack.core.regulator.IBitrateRegulator +import io.github.thibaultbee.streampack.core.regulator.controllers.IntervalBitrateRegulatorController +import io.github.thibaultbee.streampack.core.regulator.controllers.IntervalBitrateRegulatorController.Companion.DEFAULT_POLLING_TIME import io.github.thibaultbee.streampack.ext.srt.regulator.DummySrtBitrateRegulator import io.github.thibaultbee.streampack.ext.srt.regulator.SrtBitrateRegulator +import kotlin.time.Duration /** - * A [SimpleBitrateRegulatorController.Factory] for [SrtBitrateRegulator]. + * A [IntervalBitrateRegulatorController.Factory] for [SrtBitrateRegulator]. * * @param bitrateRegulatorFactory the [SrtBitrateRegulator.Factory] implementation. Use it to make your own bitrate regulator. * @param bitrateRegulatorConfig bitrate regulator configuration - * @param pollingTimeInMs delay between each call to [IBitrateRegulator.update] + * @param pollingTime delay between each call to [IBitrateRegulator.update] * - * @see SimpleBitrateRegulatorController.Factory + * @see IntervalBitrateRegulatorController.Factory * @see DummySrtBitrateRegulator.Factory */ -fun simpleSrtBitrateRegulatorControllerFactory( +fun intervalSrtBitrateRegulatorControllerFactory( bitrateRegulatorFactory: SrtBitrateRegulator.Factory = DummySrtBitrateRegulator.Factory(), bitrateRegulatorConfig: BitrateRegulatorConfig = BitrateRegulatorConfig(), - pollingTimeInMs: Long = DEFAULT_POLLING_TIME_IN_MS -) = SimpleBitrateRegulatorController.Factory( + pollingTime: Duration = DEFAULT_POLLING_TIME +) = IntervalBitrateRegulatorController.Factory( bitrateRegulatorFactory, bitrateRegulatorConfig, - pollingTimeInMs + pollingTime ) diff --git a/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/utils/SrtStatsHelper.kt b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/utils/SrtStatsHelper.kt new file mode 100644 index 000000000..e337b3575 --- /dev/null +++ b/extensions/srt/src/main/java/io/github/thibaultbee/streampack/ext/srt/utils/SrtStatsHelper.kt @@ -0,0 +1,107 @@ +/* + * 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.ext.srt.utils + +import io.github.thibaultbee.srtdroid.core.models.Stats + +object SrtStatsHelper { + val ZERO = Stats( + msTimeStamp = 0L, + pktSentTotal = 0L, + pktRecvTotal = 0L, + pktSndLossTotal = 0, + pktRcvLossTotal = 0, + pktRetransTotal = 0, + pktSentACKTotal = 0, + pktRecvACKTotal = 0, + pktSentNAKTotal = 0, + pktRecvNAKTotal = 0, + usSndDurationTotal = 0, + pktSndDropTotal = 0, + pktRcvDropTotal = 0, + pktRcvUndecryptTotal = 0, + byteSentTotal = 0, + byteRecvTotal = 0, + byteRcvLossTotal = 0, + byteRetransTotal = 0, + byteSndDropTotal = 0, + byteRcvDropTotal = 0, + byteRcvUndecryptTotal = 0, + pktSent = 0, + pktRecv = 0, + pktSndLoss = 0, + pktRcvLoss = 0, + pktRetrans = 0, + pktRcvRetrans = 0, + pktSentACK = 0, + pktRecvACK = 0, + pktSentNAK = 0, + pktRecvNAK = 0, + mbpsSendRate = 0.0, + mbpsRecvRate = 0.0, + usSndDuration = 0, + pktReorderDistance = 0, + pktRcvAvgBelatedTime = 0.0, + pktRcvBelated = 0, + pktSndDrop = 0, + pktRcvDrop = 0, + pktRcvUndecrypt = 0, + byteSent = 0, + byteRecv = 0, + byteRcvLoss = 0, + byteRetrans = 0, + byteSndDrop = 0, + byteRcvDrop = 0, + byteRcvUndecrypt = 0, + usPktSndPeriod = 0.0, + pktFlowWindow = 0, + pktCongestionWindow = 0, + pktFlightSize = 0, + msRTT = 0.0, + mbpsBandwidth = 0.0, + byteAvailSndBuf = 0, + byteAvailRcvBuf = 0, + mbpsMaxBW = 0.0, + byteMSS = 0, + pktSndBuf = 0, + byteSndBuf = 0, + msSndBuf = 0, + msSndTsbPdDelay = 0, + pktRcvBuf = 0, + byteRcvBuf = 0, + msRcvBuf = 0, + msRcvTsbPdDelay = 0, + pktSndFilterExtraTotal = 0, + pktRcvFilterExtraTotal = 0, + pktRcvFilterSupplyTotal = 0, + pktRcvFilterLossTotal = 0, + pktSndFilterExtra = 0, + pktRcvFilterExtra = 0, + pktRcvFilterSupply = 0, + pktRcvFilterLoss = 0, + pktReorderTolerance = 0, + pktSentUniqueTotal = 0, + pktRecvUniqueTotal = 0, + byteSentUniqueTotal = 0, + byteRecvUniqueTotal = 0, + pktSentUnique = 0, + pktRecvUnique = 0, + byteSentUnique = 0, + byteRecvUnique = 0 + ) +} + + diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 2cb2b2401..feda42b74 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -27,7 +27,7 @@ kotlinxIo = "0.8.0" material = "1.13.0" mockk = "1.14.5" robolectric = "4.16" -komuxer = "0.3.4" +komuxer = "0.4.0" srtdroid = "1.9.5" junitKtx = "1.3.0" compose = "1.10.1"