feat(rtmp): write/read RTMP messages from channels

Author: ThibaultBee

rtmp/src/commonMain/kotlin/io/github/thibaultbee/krtmp/rtmp/RtmpConnectionBuilder.kt

@@ -46,12 +46,12 @@ class RtmpConnectionBuilder(val selectorManager: SelectorManager) {
      * The [urlBuilder] must use the `rtmp`, `rtmps`, `rtmpt` or `rtmpts` protocol.
      *
      * @param urlBuilder the URL to connect to
-     * @param configure the settings for the RTMP client
+     * @param settings the settings for the RTMP client
      * @param message the callback to handle RTMP client events
      */
     suspend fun connect(
         urlBuilder: URLBuilder,
-        configure: RtmpClientSettings.() -> Unit = {},
+        settings: RtmpClientSettings = RtmpClientSettings(),
         message: RtmpClientCallbackBuilder.() -> Unit = {}
     ): RtmpClient {
         urlBuilder.validateRtmp()
@@ -66,18 +66,17 @@ class RtmpConnectionBuilder(val selectorManager: SelectorManager) {
             TcpSocket(tcpSocket, urlBuilder)
         }
 
-        return connect(socket, configure, message)
+        return connect(socket, settings, message)
     }
 
     /**
      * Connects to the given [socket] and performs the RTMP handshake.
      */
     private suspend fun connect(
         socket: ISocket,
-        configure: RtmpClientSettings.() -> Unit,
+        settings: RtmpClientSettings,
         message: RtmpClientCallbackBuilder.() -> Unit
     ): RtmpClient {
-        val settings = RtmpClientSettings().apply { configure() }
         try {
             socket.clientHandshake(settings.clock)
         } catch (t: Throwable) {
@@ -103,30 +102,30 @@ class RtmpConnectionBuilder(val selectorManager: SelectorManager) {
      * Binds a new [RtmpServer] to the given [localAddress].
      *
      * @param localAddress the local address to bind to. If null, binds to a random port on all interfaces.
-     * @param configure the settings for the RTMP server
+     * @param settings the settings for the RTMP server
      * @param message the callback to handle RTMP server events
      * @return a new [RtmpServer] instance
      */
     suspend fun bind(
         localAddress: SocketAddress? = null,
-        configure: RtmpServerSettings.() -> Unit = {},
+        settings: RtmpServerSettings = RtmpServerSettings(),
         message: RtmpServerCallbackBuilder.() -> Unit = {}
     ): RtmpServer {
         val serverSocket = tcpSocketBuilder.bind(localAddress)
 
-        return bind(serverSocket, configure, message)
+        return bind(serverSocket, settings, message)
     }
 
     /**
      * Binds a new [RtmpServer] to the given [serverSocket].
      */
     private fun bind(
         serverSocket: ServerSocket,
-        settings: RtmpServerSettings.() -> Unit,
+        settings: RtmpServerSettings,
         messages: RtmpServerCallbackBuilder.() -> Unit
     ) = RtmpServer(
         serverSocket,
-        RtmpServerSettings().apply { settings() },
+        settings,
         RtmpServerCallbackBuilder().apply { messages() }
     )
 }
@@ -137,28 +136,28 @@ class RtmpConnectionBuilder(val selectorManager: SelectorManager) {
  * The [urlString] must use the `rtmp`, `rtmps`, `rtmpt` or `rtmpts` protocol.
  *
  * @param urlString the RTMP URL to connect to
- * @param configure the settings for the RTMP client
+ * @param settings the settings for the RTMP client
  * @param message the callback to handle RTMP client events
  */
 suspend fun RtmpConnectionBuilder.connect(
     urlString: String,
-    configure: RtmpClientSettings.() -> Unit = {},
+    settings: RtmpClientSettings = RtmpClientSettings(),
     message: RtmpClientCallbackBuilder.() -> Unit = {}
-) = connect(RtmpURLBuilder(urlString), configure, message)
+) = connect(RtmpURLBuilder(urlString), settings, message)
 
 /**
  * Binds a new [RtmpServer] to the given [urlString].
  *
  * The [urlString] must be in the format `tcp://host:port` or `host:port`.
  *
  * @param urlString the URL string to bind to
- * @param configure the settings for the RTMP server
+ * @param settings the settings for the RTMP server
  * @param message the callback to handle RTMP server events
  * @return a new [RtmpServer] instance
  */
 suspend fun RtmpConnectionBuilder.bind(
     urlString: String,
-    configure: RtmpServerSettings.() -> Unit = {},
+    settings: RtmpServerSettings = RtmpServerSettings(),
     message: RtmpServerCallbackBuilder.() -> Unit = {}
 ): RtmpServer {
     val url = if (urlString.startWithScheme()) {
@@ -173,5 +172,5 @@ suspend fun RtmpConnectionBuilder.bind(
             url.port
         }
     )
-    return bind(localAddress, configure, message)
+    return bind(localAddress, settings, message)
 }

rtmp/src/commonMain/kotlin/io/github/thibaultbee/krtmp/rtmp/RtmpConstants.kt

@@ -30,4 +30,11 @@ object RtmpConstants {
      * The range of valid chunk sizes for RTMP connections.
      */
     internal val chunkSizeRange = MIN_CHUNK_SIZE..MAX_CHUNK_SIZE
+
+    /**
+     * The default chunk size used for RTMP connections.
+     * This value is used when the client does not specify a chunk size during the handshake.
+     * The default value is 128 bytes, which is the minimum chunk size allowed by the RTMP protocol.
+     */
+    const val DEFAULT_CHUNK_SIZE = MIN_CHUNK_SIZE // bytes
 }

rtmp/src/commonMain/kotlin/io/github/thibaultbee/krtmp/rtmp/client/RtmpClient.kt

@@ -120,6 +120,7 @@ class RtmpClient internal constructor(
      *
      * @param metadata the on metadata to send
      * @param timestampMs the timestamp of the metadata in milliseconds (usually 0)
+     * @return the deferred that will be completed when the frame is sent
      */
     suspend fun writeSetDataFrame(metadata: Metadata, timestampMs: Int = 0) =
         connection.writeSetDataFrame(metadata, timestampMs)
@@ -133,6 +134,7 @@ class RtmpClient internal constructor(
      * @param onMetadata the on metadata to send
      * @param onMetadataSize the size of the on metadata
      * @param timestampMs the timestamp of the metadata in milliseconds (usually 0)
+     * @return the deferred that will be completed when the frame is sent
      */
     suspend fun writeSetDataFrame(
         onMetadata: RawSource,
@@ -149,6 +151,7 @@ class RtmpClient internal constructor(
      * @param source the audio frame to write
      * @param sourceSize the size of the audio frame
      * @param timestampMs the timestamp of the frame in milliseconds
+     * @return the deferred that will be completed when the frame is sent
      */
     suspend fun writeAudio(source: RawSource, sourceSize: Int, timestampMs: Int) =
         connection.writeAudio(source, sourceSize, timestampMs)
@@ -161,6 +164,7 @@ class RtmpClient internal constructor(
      * @param source the video frame to write
      * @param sourceSize the size of the video frame
      * @param timestampMs the timestamp of the frame in milliseconds
+     * @return the deferred that will be completed when the frame is sent
      */
     suspend fun writeVideo(source: RawSource, sourceSize: Int, timestampMs: Int) =
         connection.writeVideo(source, sourceSize, timestampMs)
@@ -171,6 +175,7 @@ class RtmpClient internal constructor(
      * The frame must be in the FLV format.
      *
      * @param source the frame to write
+     * @return the deferred that will be completed when the frame is sent
      */
     suspend fun write(source: Source) = connection.write(source)
 
@@ -179,20 +184,23 @@ class RtmpClient internal constructor(
      *
      * @param data the frame to write
      * @param timestampMs the timestamp of the frame in milliseconds
+     * @return the deferred that will be completed when the frame is sent
      */
     suspend fun write(data: FLVData, timestampMs: Int) = connection.write(data, timestampMs)
 
     /**
      * Writes a [FLVTag].
      *
      * @param tag the FLV tag to write
+     * @return the deferred that will be completed when the tag is sent
      */
     suspend fun write(tag: FLVTag) = connection.write(tag)
 
     /**
      * Writes a [FLVTagRawBody].
      *
      * @param rawTag the FLV tag to write
+     * @return the deferred that will be completed when the tag is sent
      */
     suspend fun write(rawTag: FLVTagRawBody) = connection.write(rawTag)
 

rtmp/src/commonMain/kotlin/io/github/thibaultbee/krtmp/rtmp/client/RtmpClientSettings.kt

@@ -26,11 +26,13 @@ import io.github.thibaultbee.krtmp.rtmp.util.RtmpClock
  * @param writeWindowAcknowledgementSize RTMP acknowledgement window size in bytes
  * @param amfVersion AMF version
  * @param clock Clock used to timestamp RTMP messages. You should use the same clock for your video and audio timestamps.
+ * @param tooLateFrameDropTimeoutInMs the timeout after which a frame will be dropped (from frame timestamps). Make sure frame timestamps are on on the same clock as [clock]. If null is provided, frames will never be dropped. Default is null
  * @param connectInfo Lambda to configure the connect command object.
  */
 class RtmpClientSettings(
     writeWindowAcknowledgementSize: Int = Int.MAX_VALUE,
     amfVersion: AmfVersion = AmfVersion.AMF0,
     clock: RtmpClock = RtmpClock.Default(),
-    var connectInfo: ConnectObjectBuilder.() -> Unit = {}
-) : RtmpSettings(writeWindowAcknowledgementSize, amfVersion, clock, false, 0L)
+    tooLateFrameDropTimeoutInMs: Long? = null,
+    val connectInfo: ConnectObjectBuilder.() -> Unit = {}
+) : RtmpSettings(writeWindowAcknowledgementSize, amfVersion, clock, tooLateFrameDropTimeoutInMs)