WebRtc Client. Add Android implementation for datachannels.

Author: zibet27

ktor-client/ktor-client-webrtc/android/src/io/ktor/client/webrtc/DataChannel.kt

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2014-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.client.webrtc
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.launch
+import org.webrtc.DataChannel
+import java.nio.ByteBuffer
+
+/**
+ * WebRtc data channel implementation for the Android platform.
+ */
+public class AndroidWebRtcDataChannel(
+    internal val nativeChannel: DataChannel,
+    private val channelInit: DataChannel.Init?,
+    private val coroutineScope: CoroutineScope,
+    options: WebRtcDataChannelOptions
+) : WebRtcDataChannel(options) {
+
+    override val id: Int
+        get() = nativeChannel.id()
+
+    override val label: String
+        get() = nativeChannel.label()
+
+    override val state: WebRtc.DataChannelState
+        get() = nativeChannel.state().toKtor()
+
+    override val bufferedAmount: Long
+        get() = nativeChannel.bufferedAmount()
+
+    override var bufferedAmountLowThreshold: Long = 0
+        private set
+
+    override val maxPacketLifeTime: Int?
+        get() = channelInit?.maxRetransmitTimeMs
+
+    override val maxRetransmits: Int?
+        get() = channelInit?.maxRetransmits
+
+    override val negotiated: Boolean
+        get() = channelInit?.negotiated ?: error("Can't retrieve negotiated state on Android")
+
+    override val ordered: Boolean
+        get() = channelInit?.ordered ?: error("Can't retrieve ordered state on Android")
+
+    override val protocol: String
+        get() = channelInit?.protocol ?: error("Protocol is not supported in WebRTC")
+
+    private fun checkStatus() {
+        if (state == WebRtc.DataChannelState.CLOSING || state == WebRtc.DataChannelState.CLOSED) {
+            error("Data channel is closed.")
+        }
+    }
+
+    override fun send(text: String) {
+        checkStatus()
+        val buffer = DataChannel.Buffer(Charsets.UTF_8.encode(text), false)
+        nativeChannel.send(buffer)
+    }
+
+    override fun send(bytes: ByteArray) {
+        checkStatus()
+        val buffer = DataChannel.Buffer(ByteBuffer.wrap(bytes), true)
+        nativeChannel.send(buffer)
+    }
+
+    override fun setBufferedAmountLowThreshold(threshold: Long) {
+        bufferedAmountLowThreshold = threshold
+    }
+
+    override fun close() {
+        nativeChannel.close()
+    }
+
+    internal fun setupEvents(eventsEmitter: WebRtcConnectionEventsEmitter) {
+        nativeChannel.registerObserver(object : DataChannel.Observer {
+            override fun onStateChange() {
+                coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) {
+                    val event = when (state) {
+                        WebRtc.DataChannelState.CONNECTING -> null
+                        WebRtc.DataChannelState.OPEN -> DataChannelEvent.Open(this@AndroidWebRtcDataChannel)
+                        WebRtc.DataChannelState.CLOSING -> DataChannelEvent.Closing(this@AndroidWebRtcDataChannel)
+                        WebRtc.DataChannelState.CLOSED -> {
+                            stopReceivingMessages()
+                            DataChannelEvent.Closed(this@AndroidWebRtcDataChannel)
+                        }
+                    }
+                    event?.let { eventsEmitter.emitDataChannelEvent(it) }
+                }
+            }
+
+            override fun onMessage(buffer: DataChannel.Buffer?) {
+                // This coroutine should start immediately because the protocol relies on the messages order
+                coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) {
+                    if (buffer == null) {
+                        return@launch
+                    }
+                    val message = when (buffer.binary) {
+                        true -> {
+                            val data = ByteArray(buffer.data.remaining()).apply { buffer.data.get(this) }
+                            Message.Binary(data)
+                        }
+
+                        false -> Message.Text(Charsets.UTF_8.decode(buffer.data).toString())
+                    }
+                    emitMessage(message)
+                }
+            }
+
+            override fun onBufferedAmountChange(previousAmount: Long) {
+                coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) {
+                    if (previousAmount > bufferedAmountLowThreshold && bufferedAmountLowThreshold > bufferedAmount) {
+                        return@launch
+                    }
+                    val event = DataChannelEvent.BufferedAmountLow(this@AndroidWebRtcDataChannel)
+                    eventsEmitter.emitDataChannelEvent(event)
+                }
+            }
+        })
+    }
+}
+
+/**
+ * Returns implementation of the data channel that is used under the hood. Use it with caution.
+ */
+public fun WebRtcDataChannel.getNative(): DataChannel {
+    val channel = (this as? AndroidWebRtcDataChannel) ?: error("Wrong data channel implementation.")
+    return channel.nativeChannel
+}

ktor-client/ktor-client-webrtc/android/src/io/ktor/client/webrtc/Engine.kt

@@ -54,7 +54,7 @@ public class AndroidWebRtcEngine(
             iceTransportsType = config.iceTransportPolicy.toNative()
         }
 
-        val coroutineContext = createConnectionContext(config.coroutinesContext)
+        val coroutineContext = createConnectionContext(config.coroutineContext)
         return AndroidWebRtcPeerConnection(coroutineContext, config).initialize { observer ->
             getLocalFactory().createPeerConnection(rtcConfig, observer)
         }

ktor-client/ktor-client-webrtc/android/src/io/ktor/client/webrtc/PeerConnection.kt

@@ -5,6 +5,8 @@
 package io.ktor.client.webrtc
 
 import io.ktor.client.webrtc.media.*
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.launch
 import org.webrtc.*
 import org.webrtc.PeerConnection.Observer
 import kotlin.coroutines.CoroutineContext
@@ -25,7 +27,7 @@ public class AndroidWebRtcPeerConnection(
         if (this::peerConnection.isInitialized.not()) {
             cont.resume(emptyList())
         }
-        peerConnection.getStats { cont.resume(it.toCommon()) }
+        peerConnection.getStats { cont.resume(it.toKtor()) }
     }
 
     // helper method to break a dependency cycle (PeerConnection -> PeerConnectionFactory -> Observer)
@@ -44,84 +46,120 @@ public class AndroidWebRtcPeerConnection(
         if (hasVideo) mandatory.add(MediaConstraints.KeyValuePair("OfferToReceiveVideo", "true"))
     }
 
+    private inline fun runInConnectionScope(crossinline block: () -> Unit) {
+        // Runs a `block` in the coroutine of the peer connection not to lose possible exceptions.
+        // We are already running on the special thread, so extra dispatching is not required.
+        // Moreover, dispatching the coroutine on another thread could break the internal `org.webrtc` logic.
+        // For instance, it silently breaks registering a data channel observer.
+        coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) { block() }
+    }
+
     private fun createObserver() = object : Observer {
-        override fun onIceCandidate(candidate: IceCandidate?) {
-            if (candidate == null) return
-            events.emitIceCandidate(candidate.toCommon())
+        override fun onIceCandidate(candidate: IceCandidate?) = runInConnectionScope {
+            if (candidate == null) return@runInConnectionScope
+            events.emitIceCandidate(candidate.toKtor())
         }
 
         override fun onIceCandidatesRemoved(candidates: Array<out IceCandidate>?) = Unit
 
-        override fun onAddTrack(receiver: RtpReceiver?, mediaStreams: Array<out MediaStream>?) {
-            if (receiver == null) return
+        override fun onAddTrack(receiver: RtpReceiver?, mediaStreams: Array<out MediaStream>?) = runInConnectionScope {
+            if (receiver == null) return@runInConnectionScope
             receiver.track()?.let {
                 events.emitAddTrack(AndroidMediaTrack.from(it))
             }
         }
 
-        override fun onRemoveTrack(receiver: RtpReceiver?) {
-            if (receiver == null) return
+        override fun onRemoveTrack(receiver: RtpReceiver?) = runInConnectionScope {
+            if (receiver == null) return@runInConnectionScope
             receiver.track()?.let {
                 events.emitRemoveTrack(AndroidMediaTrack.from(it))
             }
         }
 
-        override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
-            val commonState = newState.toCommon() ?: return
+        override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) = runInConnectionScope {
+            val commonState = newState.toKtor() ?: return@runInConnectionScope
             events.emitIceConnectionStateChange(commonState)
         }
 
-        override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
-            val commonState = newState.toCommon() ?: return
+        override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) = runInConnectionScope {
+            val commonState = newState.toKtor() ?: return@runInConnectionScope
             events.emitConnectionStateChange(commonState)
         }
 
-        override fun onSignalingChange(newState: PeerConnection.SignalingState?) {
-            val commonState = newState.toCommon() ?: return
+        override fun onSignalingChange(newState: PeerConnection.SignalingState?) = runInConnectionScope {
+            val commonState = newState.toKtor() ?: return@runInConnectionScope
             events.emitSignalingStateChange(commonState)
         }
 
-        override fun onIceGatheringChange(newState: PeerConnection.IceGatheringState?) {
-            val commonState = newState.toCommon() ?: return
+        override fun onIceGatheringChange(newState: PeerConnection.IceGatheringState?) = runInConnectionScope {
+            val commonState = newState.toKtor() ?: return@runInConnectionScope
             events.emitIceGatheringStateChange(commonState)
         }
 
         override fun onIceConnectionReceivingChange(receiving: Boolean) = Unit
-        override fun onRenegotiationNeeded(): Unit = events.emitNegotiationNeeded()
+        override fun onRenegotiationNeeded() = runInConnectionScope {
+            events.emitNegotiationNeeded()
+        }
 
         // we omit streams and operate with tracks instead
         override fun onAddStream(p0: MediaStream?) = Unit
         override fun onRemoveStream(p0: MediaStream?) = Unit
 
-        // #TODO: implement data channels
-        override fun onDataChannel(dataChannel: DataChannel?) = Unit
+        override fun onDataChannel(dataChannel: DataChannel?) = runInConnectionScope {
+            if (dataChannel == null) return@runInConnectionScope
+            val channel = AndroidWebRtcDataChannel(
+                nativeChannel = dataChannel,
+                channelInit = null,
+                coroutineScope = coroutineScope,
+                options = WebRtcDataChannelOptions()
+            )
+            channel.setupEvents(events)
+        }
     }
 
     override val localDescription: WebRtc.SessionDescription?
-        get() = peerConnection.localDescription?.toCommon()
+        get() = peerConnection.localDescription?.toKtor()
 
     override val remoteDescription: WebRtc.SessionDescription?
-        get() = peerConnection.remoteDescription?.toCommon()
+        get() = peerConnection.remoteDescription?.toKtor()
 
     override suspend fun createOffer(): WebRtc.SessionDescription {
         val offer = suspendCoroutine { cont ->
             peerConnection.createOffer(cont.resumeAfterSdpCreate(), offerConstraints())
         }
-        return offer.toCommon()
+        return offer.toKtor()
     }
 
     override suspend fun createAnswer(): WebRtc.SessionDescription {
         val answer = suspendCoroutine { cont ->
             peerConnection.createAnswer(cont.resumeAfterSdpCreate(), offerConstraints())
         }
-        return answer.toCommon()
+        return answer.toKtor()
     }
 
     override suspend fun createDataChannel(
         label: String,
         options: WebRtcDataChannelOptions.() -> Unit
     ): WebRtcDataChannel {
-        TODO("Not yet implemented")
+        val options = WebRtcDataChannelOptions().apply(options)
+        val channelInit = DataChannel.Init().apply {
+            if (options.id != null) {
+                id = options.id!!
+            }
+            if (options.maxRetransmits != null) {
+                maxRetransmits = options.maxRetransmits!!
+            }
+            if (options.maxPacketLifeTime != null) {
+                maxRetransmitTimeMs = options.maxPacketLifeTime?.inWholeMilliseconds?.toInt()!!
+            }
+            ordered = options.ordered
+            protocol = options.protocol
+            negotiated = options.negotiated
+        }
+        val nativeChannel = peerConnection.createDataChannel(label, channelInit)
+        return AndroidWebRtcDataChannel(nativeChannel, channelInit, coroutineScope, options).apply {
+            setupEvents(events)
+        }
     }
 
     override suspend fun setLocalDescription(description: WebRtc.SessionDescription) {

ktor-client/ktor-client-webrtc/android/src/io/ktor/client/webrtc/Utils.kt

@@ -4,11 +4,7 @@
 
 package io.ktor.client.webrtc
 
-import org.webrtc.IceCandidate
-import org.webrtc.PeerConnection
-import org.webrtc.RTCStatsReport
-import org.webrtc.SdpObserver
-import org.webrtc.SessionDescription
+import org.webrtc.*
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
@@ -41,7 +37,7 @@ internal fun WebRtc.RtcpMuxPolicy.toNative(): PeerConnection.RtcpMuxPolicy = whe
     WebRtc.RtcpMuxPolicy.REQUIRE -> PeerConnection.RtcpMuxPolicy.REQUIRE
 }
 
-internal fun PeerConnection.IceConnectionState?.toCommon(): WebRtc.IceConnectionState? = when (this) {
+internal fun PeerConnection.IceConnectionState?.toKtor(): WebRtc.IceConnectionState? = when (this) {
     PeerConnection.IceConnectionState.NEW -> WebRtc.IceConnectionState.NEW
     PeerConnection.IceConnectionState.FAILED -> WebRtc.IceConnectionState.FAILED
     PeerConnection.IceConnectionState.CLOSED -> WebRtc.IceConnectionState.CLOSED
@@ -52,7 +48,7 @@ internal fun PeerConnection.IceConnectionState?.toCommon(): WebRtc.IceConnection
     else -> null
 }
 
-internal fun PeerConnection.PeerConnectionState?.toCommon(): WebRtc.ConnectionState? = when (this) {
+internal fun PeerConnection.PeerConnectionState?.toKtor(): WebRtc.ConnectionState? = when (this) {
     PeerConnection.PeerConnectionState.NEW -> WebRtc.ConnectionState.NEW
     PeerConnection.PeerConnectionState.CLOSED -> WebRtc.ConnectionState.CLOSED
     PeerConnection.PeerConnectionState.CONNECTED -> WebRtc.ConnectionState.CONNECTED
@@ -62,7 +58,7 @@ internal fun PeerConnection.PeerConnectionState?.toCommon(): WebRtc.ConnectionSt
     else -> null
 }
 
-internal fun PeerConnection.SignalingState?.toCommon(): WebRtc.SignalingState? = when (this) {
+internal fun PeerConnection.SignalingState?.toKtor(): WebRtc.SignalingState? = when (this) {
     PeerConnection.SignalingState.STABLE -> WebRtc.SignalingState.STABLE
     PeerConnection.SignalingState.CLOSED -> WebRtc.SignalingState.CLOSED
     PeerConnection.SignalingState.HAVE_LOCAL_OFFER -> WebRtc.SignalingState.HAVE_LOCAL_OFFER
@@ -72,14 +68,21 @@ internal fun PeerConnection.SignalingState?.toCommon(): WebRtc.SignalingState? =
     else -> null
 }
 
-internal fun PeerConnection.IceGatheringState?.toCommon(): WebRtc.IceGatheringState? = when (this) {
+internal fun PeerConnection.IceGatheringState?.toKtor(): WebRtc.IceGatheringState? = when (this) {
     PeerConnection.IceGatheringState.NEW -> WebRtc.IceGatheringState.NEW
     PeerConnection.IceGatheringState.COMPLETE -> WebRtc.IceGatheringState.COMPLETE
     PeerConnection.IceGatheringState.GATHERING -> WebRtc.IceGatheringState.GATHERING
     else -> null
 }
 
-internal fun SessionDescription.toCommon(): WebRtc.SessionDescription {
+internal fun DataChannel.State.toKtor(): WebRtc.DataChannelState = when (this) {
+    DataChannel.State.CONNECTING -> WebRtc.DataChannelState.CONNECTING
+    DataChannel.State.OPEN -> WebRtc.DataChannelState.OPEN
+    DataChannel.State.CLOSING -> WebRtc.DataChannelState.CLOSING
+    DataChannel.State.CLOSED -> WebRtc.DataChannelState.CLOSED
+}
+
+internal fun SessionDescription.toKtor(): WebRtc.SessionDescription {
     return WebRtc.SessionDescription(
         when (requireNotNull(type)) {
             SessionDescription.Type.OFFER -> WebRtc.SessionDescriptionType.OFFER
@@ -91,7 +94,7 @@ internal fun SessionDescription.toCommon(): WebRtc.SessionDescription {
     )
 }
 
-internal fun RTCStatsReport.toCommon(): List<WebRtc.Stats> = statsMap.values.map {
+internal fun RTCStatsReport.toKtor(): List<WebRtc.Stats> = statsMap.values.map {
     WebRtc.Stats(
         it.id,
         it.type,
@@ -100,7 +103,7 @@ internal fun RTCStatsReport.toCommon(): List<WebRtc.Stats> = statsMap.values.map
     )
 }
 
-internal fun IceCandidate.toCommon(): WebRtc.IceCandidate = WebRtc.IceCandidate(
+internal fun IceCandidate.toKtor(): WebRtc.IceCandidate = WebRtc.IceCandidate(
     candidate = sdp,
     sdpMid = sdpMid,
     sdpMLineIndex = sdpMLineIndex