WebRTC Client. Create peer connection in the init block. (#5133)

Author: zibet27

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

@@ -54,7 +54,7 @@ public class AndroidWebRtcEngine(
         }
 
         val coroutineContext = createConnectionContext(config.exceptionHandler)
-        return AndroidWebRtcPeerConnection(coroutineContext, config).initialize { observer ->
+        return AndroidWebRtcPeerConnection(coroutineContext, config) { observer ->
             localFactory.createPeerConnection(rtcConfig, observer)
         }
     }

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

@@ -5,8 +5,6 @@
 package io.ktor.client.webrtc
 
 import io.ktor.client.webrtc.media.*
-import kotlinx.coroutines.CoroutineStart
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
 import org.webrtc.*
 import org.webrtc.PeerConnection.Observer
@@ -16,29 +14,25 @@ import kotlin.coroutines.resumeWithException
 
 public class AndroidWebRtcPeerConnection(
     coroutineContext: CoroutineContext,
-    config: WebRtcConnectionConfig
+    config: WebRtcConnectionConfig,
+    createConnection: (Observer) -> PeerConnection?
 ) : WebRtcPeerConnection(coroutineContext, config) {
-    internal lateinit var peerConnection: PeerConnection
+    internal val peerConnection: PeerConnection
+
+    init {
+        peerConnection = createConnection(createObserver())
+            ?: error(
+                "Failed to create peer connection. On the Android platform it is usually caused by invalid configuration. For instance, missing turn server username."
+            )
+    }
 
     // remember RTP senders because method PeerConnection.getSenders() disposes all returned senders
     private val rtpSenders = arrayListOf<AndroidRtpSender>()
 
     override suspend fun getStatistics(): List<WebRtc.Stats> = suspendCancellableCoroutine { cont ->
-        if (this::peerConnection.isInitialized.not()) {
-            cont.resume(emptyList())
-            return@suspendCancellableCoroutine
-        }
         peerConnection.getStats { cont.resume(it.toKtor()) }
     }
 
-    // helper method to break a dependency cycle (PeerConnection -> PeerConnectionFactory -> Observer)
-    public fun initialize(block: (Observer) -> PeerConnection?): AndroidWebRtcPeerConnection {
-        peerConnection = block(createObserver()) ?: error(
-            "Failed to create peer connection. On the Android platform it is usually caused by invalid configuration. For instance, missing turn server username."
-        )
-        return this
-    }
-
     private val hasVideo get() = rtpSenders.any { it.track?.kind == WebRtcMedia.TrackType.VIDEO }
     private val hasAudio get() = rtpSenders.any { it.track?.kind == WebRtcMedia.TrackType.AUDIO }
 
@@ -47,14 +41,6 @@ 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?) = runInConnectionScope {
             val candidate = candidate?.toKtor() ?: return@runInConnectionScope

ktor-client/ktor-client-webrtc/api/android/ktor-client-webrtc.api

@@ -72,7 +72,7 @@ public final class io/ktor/client/webrtc/AndroidWebRtcEngineConfig : io/ktor/cli
 }
 
 public final class io/ktor/client/webrtc/AndroidWebRtcPeerConnection : io/ktor/client/webrtc/WebRtcPeerConnection {
-	public fun <init> (Lkotlin/coroutines/CoroutineContext;Lio/ktor/client/webrtc/WebRtcConnectionConfig;)V
+	public fun <init> (Lkotlin/coroutines/CoroutineContext;Lio/ktor/client/webrtc/WebRtcConnectionConfig;Lkotlin/jvm/functions/Function1;)V
 	public fun addIceCandidate (Lio/ktor/client/webrtc/WebRtc$IceCandidate;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun addTrack (Lio/ktor/client/webrtc/WebRtcMedia$Track;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun close ()V
@@ -82,7 +82,6 @@ public final class io/ktor/client/webrtc/AndroidWebRtcPeerConnection : io/ktor/c
 	public fun getLocalDescription ()Lio/ktor/client/webrtc/WebRtc$SessionDescription;
 	public fun getRemoteDescription ()Lio/ktor/client/webrtc/WebRtc$SessionDescription;
 	public fun getStatistics (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public final fun initialize (Lkotlin/jvm/functions/Function1;)Lio/ktor/client/webrtc/AndroidWebRtcPeerConnection;
 	public fun removeTrack (Lio/ktor/client/webrtc/WebRtc$RtpSender;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun removeTrack (Lio/ktor/client/webrtc/WebRtcMedia$Track;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun restartIce ()V

ktor-client/ktor-client-webrtc/api/ktor-client-webrtc.klib.api

@@ -969,15 +969,14 @@ final class io.ktor.client.webrtc/IosRtpSender : io.ktor.client.webrtc/WebRtc.Rt
 
 // Targets: [apple]
 final class io.ktor.client.webrtc/IosWebRtcConnection : io.ktor.client.webrtc/WebRtcPeerConnection { // io.ktor.client.webrtc/IosWebRtcConnection|null[0]
-    constructor <init>(kotlin.coroutines/CoroutineContext, io.ktor.client.webrtc/WebRtcConnectionConfig) // io.ktor.client.webrtc/IosWebRtcConnection.<init>|<init>(kotlin.coroutines.CoroutineContext;io.ktor.client.webrtc.WebRtcConnectionConfig){}[0]
+    constructor <init>(kotlin.coroutines/CoroutineContext, io.ktor.client.webrtc/WebRtcConnectionConfig, kotlin/Function1<WebRTC/RTCPeerConnectionDelegateProtocol, WebRTC/RTCPeerConnection?>) // io.ktor.client.webrtc/IosWebRtcConnection.<init>|<init>(kotlin.coroutines.CoroutineContext;io.ktor.client.webrtc.WebRtcConnectionConfig;kotlin.Function1<WebRTC.RTCPeerConnectionDelegateProtocol,WebRTC.RTCPeerConnection?>){}[0]
 
     final val localDescription // io.ktor.client.webrtc/IosWebRtcConnection.localDescription|{}localDescription[0]
         final fun <get-localDescription>(): io.ktor.client.webrtc/WebRtc.SessionDescription? // io.ktor.client.webrtc/IosWebRtcConnection.localDescription.<get-localDescription>|<get-localDescription>(){}[0]
     final val remoteDescription // io.ktor.client.webrtc/IosWebRtcConnection.remoteDescription|{}remoteDescription[0]
         final fun <get-remoteDescription>(): io.ktor.client.webrtc/WebRtc.SessionDescription? // io.ktor.client.webrtc/IosWebRtcConnection.remoteDescription.<get-remoteDescription>|<get-remoteDescription>(){}[0]
 
     final fun close() // io.ktor.client.webrtc/IosWebRtcConnection.close|close(){}[0]
-    final fun initialize(kotlin/Function1<WebRTC/RTCPeerConnectionDelegateProtocol, WebRTC/RTCPeerConnection?>): io.ktor.client.webrtc/IosWebRtcConnection // io.ktor.client.webrtc/IosWebRtcConnection.initialize|initialize(kotlin.Function1<WebRTC.RTCPeerConnectionDelegateProtocol,WebRTC.RTCPeerConnection?>){}[0]
     final fun restartIce() // io.ktor.client.webrtc/IosWebRtcConnection.restartIce|restartIce(){}[0]
     final suspend fun addIceCandidate(io.ktor.client.webrtc/WebRtc.IceCandidate) // io.ktor.client.webrtc/IosWebRtcConnection.addIceCandidate|addIceCandidate(io.ktor.client.webrtc.WebRtc.IceCandidate){}[0]
     final suspend fun addTrack(io.ktor.client.webrtc/WebRtcMedia.Track): io.ktor.client.webrtc/WebRtc.RtpSender // io.ktor.client.webrtc/IosWebRtcConnection.addTrack|addTrack(io.ktor.client.webrtc.WebRtcMedia.Track){}[0]

ktor-client/ktor-client-webrtc/common/src/io/ktor/client/webrtc/WebRtcPeerConnection.kt

@@ -6,6 +6,7 @@ package io.ktor.client.webrtc
 
 import io.ktor.utils.io.core.*
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
@@ -125,6 +126,14 @@ public abstract class WebRtcPeerConnection private constructor(
         iceGatheringState.first { it == WebRtc.IceGatheringState.COMPLETE }
     }
 
+    /**
+     * Runs a [block] in the coroutine scope of the peer connection without extra dispatching.
+     * This should be used to run some background tasks without losing thrown exceptions.
+     */
+    internal inline fun runInConnectionScope(crossinline block: () -> Unit) {
+        coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) { block() }
+    }
+
     override fun close() {
         coroutineScope.cancel()
     }

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

@@ -39,9 +39,9 @@ public class IosWebRtcEngine(
 ) : WebRtcEngineBase("ios-webrtc", config), MediaTrackFactory by mediaTrackFactory {
 
     private val localFactory: RTCPeerConnectionFactory
-        get() {
-            return config.rtcFactory ?: (mediaTrackFactory as IosMediaDevices).peerConnectionFactory
-        }
+        get() = config.rtcFactory
+            ?: (mediaTrackFactory as? IosMediaDevices)?.peerConnectionFactory
+            ?: error("Please specify custom rtcFactory for custom MediaTrackFactory")
 
     private fun WebRtc.IceServer.toIos(): RTCIceServer {
         return RTCIceServer(listOf(urls), username, credential)
@@ -59,7 +59,7 @@ public class IosWebRtcEngine(
         }
 
         val coroutineContext = createConnectionContext(config.exceptionHandler)
-        return IosWebRtcConnection(coroutineContext, config).initialize { delegate ->
+        return IosWebRtcConnection(coroutineContext, config) { delegate ->
             localFactory.peerConnectionWithConfiguration(
                 constraints = RTCMediaConstraints(),
                 configuration = rtcConfig,

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

@@ -8,8 +8,6 @@ import WebRTC.*
 import io.ktor.client.webrtc.media.*
 import kotlinx.cinterop.ExperimentalForeignApi
 import kotlinx.cinterop.ObjCSignatureOverride
-import kotlinx.coroutines.CoroutineStart
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
 import platform.darwin.NSObject
 import kotlin.coroutines.CoroutineContext
@@ -21,44 +19,26 @@ import kotlin.coroutines.resumeWithException
  *
  * @param coroutineContext coroutine context used to deliver connection callbacks.
  * @param config configuration describing ICE servers, media constraints, and other connection options.
+ * @param createConnection factory function that creates a native peer connection instance using a provided delegate.
  */
 @OptIn(ExperimentalForeignApi::class)
 public class IosWebRtcConnection(
     coroutineContext: CoroutineContext,
-    config: WebRtcConnectionConfig
+    config: WebRtcConnectionConfig,
+    createConnection: (RTCPeerConnectionDelegateProtocol) -> RTCPeerConnection?
 ) : WebRtcPeerConnection(coroutineContext, config) {
-    internal lateinit var peerConnection: RTCPeerConnection
+    internal val peerConnection: RTCPeerConnection
+
+    init {
+        peerConnection = createConnection(createDelegate()) ?: error("Failed to create peer connection.")
+    }
 
     override suspend fun getStatistics(): List<WebRtc.Stats> = suspendCancellableCoroutine { cont ->
-        if (!this::peerConnection.isInitialized) {
-            cont.resume(emptyList())
-            return@suspendCancellableCoroutine
-        }
         peerConnection.statisticsWithCompletionHandler { stats ->
             cont.resume(stats?.toKtor() ?: emptyList())
         }
     }
 
-    /**
-     * Finishes constructing the underlying native peer connection by invoking [block] with the created delegate.
-     *
-     * @param block factory that receives the delegate and returns a configured [RTCPeerConnection].
-     * @return this connection instance once the native peer connection has been installed.
-     * @throws IllegalStateException if called more than once or if the native connection cannot be created.
-     */
-    public fun initialize(block: (RTCPeerConnectionDelegateProtocol) -> RTCPeerConnection?): IosWebRtcConnection {
-        if (this::peerConnection.isInitialized) {
-            error("Peer connection has been already initialized.")
-        }
-        peerConnection = block(createDelegate()) ?: error("Failed to create peer connection.")
-        return this
-    }
-
-    private inline fun runInConnectionScope(crossinline block: () -> Unit) {
-        // Runs a `block` in the coroutine of the peer connection not to lose possible exceptions.
-        coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) { block() }
-    }
-
     private fun createDelegate() = object : RTCPeerConnectionDelegateProtocol, NSObject() {
         override fun peerConnection(
             peerConnection: RTCPeerConnection,

ktor-client/ktor-client-webrtc/ktor-client-webrtc-rs/common/src/io/ktor/client/webrtc/rs/Connection.kt

@@ -18,12 +18,6 @@ public class RustWebRtcConnection(
     config: WebRtcConnectionConfig
 ) : WebRtcPeerConnection(coroutineContext, config) {
 
-    private fun runInConnectionScope(block: suspend () -> Unit) {
-        coroutineScope.launch(start = CoroutineStart.UNDISPATCHED) {
-            block()
-        }
-    }
-
     init {
         inner.registerObserver(object : PeerConnectionObserver {
             override fun onConnectionStateChange(state: ConnectionState) = runInConnectionScope {