Seamless mobile real-time communication with WebRTC

The WebRTC project allows developers to build strong voice and video communication solutions by facilitating the transmission of data between peers.
During this talk, we will discover the WebRTC project, its design, and how to implement a simple mobile app in a Kotlin Multiplatform project.

Transcript

1. Friday, June 7, 2024 droidcon San Francisco Seamless mobile real-time communication with WebRTC Renaud Mathieu
2. Renaud Mathieu • Freelance • Based in Paris, France 🥖🍷 • Member of the Paris Android User Group • Proud of being here today with you About me renaudmathieu.com
3. Web Real-Time Communication
4. 📹 Video 🔈 Audio 📦 Data
5. 📜 History
6. 2009 History Timeline
7. Early Beginnings 2009 History Timeline
8. Early Beginnings 2009 2011 Development & Standardization History Timeline
9. Early Beginnings 2009 2011 Development & Standardization Browser Adoption 2012-2013 History Timeline
10. Early Beginnings 2009 2011 Development & Standardization Browser Adoption 2012-2013 Industry Adoption 2014-2015 History Timeline
11. Early Beginnings 2009 2011 Development & Standardization Browser Adoption 2012-2013 Industry Adoption 2014-2015 2018 WebRTC 1.0 History Timeline
12. Early Beginnings 2009 2011 Development & Standardization Browser Adoption 2012-2013 Industry Adoption 2014-2015 2018 WebRTC 1.0 2020-Present Ubiquity History Timeline
13. 📐 Architecture
14. Client A Client C Client B 🕸 Mesh Number of edges= N×(N−1) / 2
15. Server Client A Client C Client B 🔗 Multipoint Conference Unit
16. Server Client A Client C Client B 📦 📦 📦 🔗 Multipoint Conference Unit
17. Server Client A Client B 👥 WebRTC 👋🪪 🤙🪪 💁 💁 🤝
18. 👥 WebRTC 1.Signaling Phase 2. Peer-to-Peer Connection Phase
19. ICE (Interactive Connectivity Establishment) 👥 WebRTC
20. STUN (Session Traversal Utilities for NAT) ICE (Interactive Connectivity Establishment) 👥 WebRTC
21. TURN (Traversal Using Relays around NAT) 👥 WebRTC STUN (Session Traversal Utilities for NAT) ICE (Interactive Connectivity Establishment)
22. Turn Server STUN Server NAT NAT Client A Client B STUN Server Signaling Server Summary
23. Additional informations • WebRTC transmits data using UDP • No Standard regarding to the Signaling Server • 🔎 chrome://webrtc-internals
24. 🧑💻 Signaling Server
27. https://start.ktor.io
28. import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable @Serializable data class WsPacket( @SerialName("id") val id: WsMessage, @SerialName("data") val data: String, )
29. import kotlinx.serialization.Serializable @Serializable enum class WsMessage { WsError, WsIceCandidates, WsIceServers, WsInit, WsSDPAnswer, WsSDPOffer, }
30. class Connection( val session: DefaultWebSocketSession ) { companion object { val lastId = AtomicInteger(0) } val name = "client${lastId.getAndIncrement()}" }
31. @OptIn(InternalAPI :: class) fun Application.configureSockets() { install(WebSockets) { pingPeriod = Duration.ofSeconds(15) timeout = Duration.ofSeconds(15) maxFrameSize = Long.MAX_VALUE masking = false contentConverter = KotlinxWebsocketSerializationConverter(Json) } val connections = Collections.synchronizedSet<Connection?>(LinkedHashSet()) routing { webSocket("/connect") // Let’s GO } } }
32. @OptIn(InternalAPI :: class) fun Application.configureSockets() { install(WebSockets) { pingPeriod = Duration.ofSeconds(15) timeout = Duration.ofSeconds(15) maxFrameSize = Long.MAX_VALUE masking = false contentConverter = KotlinxWebsocketSerializationConverter(Json) } val connections = Collections.synchronizedSet<Connection?>(LinkedHashSet()) routing { webSocket("/connect") // Let’s GO } } }
33. @OptIn(InternalAPI :: class) fun Application.configureSockets() { install(WebSockets) { pingPeriod = Duration.ofSeconds(15) timeout = Duration.ofSeconds(15) maxFrameSize = Long.MAX_VALUE masking = false contentConverter = KotlinxWebsocketSerializationConverter(Json) } val connections = Collections.synchronizedSet<Connection?>(LinkedHashSet()) routing { webSocket("/connect") // Let’s GO } } }
34. webSocket("/connect") { // Register this new client val thisConnection = Connection(this) connections += thisConnection try { while (true) { val wsPacket = receiveDeserialized<WsPacket>() when (wsPacket.id) { WsMessage.WsInit, WsMessage.WsError, WsMessage.WsIceCandidates, WsMessage.WsIceServers, WsMessage.WsSDPAnswer, WsMessage.WsSDPOffer -> { // DO SOMETHING } } } } catch (e: Exception) { println(e.localizedMessage) } finally { println("Removing $thisConnection!") connections -= thisConnection } }
35. webSocket("/connect") { // Register this new client val thisConnection = Connection(this) connections += thisConnection try { while (true) { val wsPacket = receiveDeserialized<WsPacket>() when (wsPacket.id) { WsMessage.WsInit, WsMessage.WsError, WsMessage.WsIceCandidates, WsMessage.WsIceServers, WsMessage.WsSDPAnswer, WsMessage.WsSDPOffer -> { // DO SOMETHING } } } } catch (e: Exception) { println(e.localizedMessage) } finally { println("Removing $thisConnection!") connections -= thisConnection } }
36. webSocket("/connect") { // Register this new client val thisConnection = Connection(this) connections += thisConnection try { while (true) { val wsPacket = receiveDeserialized<WsPacket>() when (wsPacket.id) { WsMessage.WsInit, WsMessage.WsError, WsMessage.WsIceCandidates, WsMessage.WsIceServers, WsMessage.WsSDPAnswer, WsMessage.WsSDPOffer -> { // DO SOMETHING } } } } catch (e: Exception) { println(e.localizedMessage) } finally { println("Removing $thisConnection!") connections -= thisConnection } }
37. WsMessage.WsIceCandidates, WsMessage.WsIceServers, WsMessage.WsSDPAnswer, WsMessage.WsSDPOffer -> { connections .filter { connection -> connection.name != thisConnection.name } .forEach { it.session.send(wsPacket.data) } }
38. <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>WebRTC Simple Example </ title> </ head> <body> <h1>WebRTC Connection </ h1> <video id="localVideo" autoplay playsinline> </ video> <video id="remoteVideo" autoplay playsinline> </ video> <button id="startCallButton">Start Call </ button> <script> const signalingServerUrl = "ws: // localhost:8080/connect"; const signalingSocket = new WebSocket(signalingServerUrl); let localPeerConnection; let remotePeerConnection; let localStream; signalingSocket.onmessage = async (message) => { const data = JSON.parse(message.data); if (data.offer) { await handleOffer(data.offer); } else if (data.answer) {
39. ▶︎
40. 🧑💻 Android
46. 💡States
47. WebSocketDataSource
48. WebSocketDataSource WebRTCDataSource
49. ConnectionRepository WebSocketDataSource WebRTCDataSource
50. ConnectionRepository WebSocketDataSource WebRTCDataSource ViewModel
51. WebSocketDataSource
52. class WebSocketDataSource { private var session: DefaultClientWebSocketSession? = null private var pingJob: Job = SupervisorJob() private val client = HttpClient { install(Logging) { level = LogLevel.ALL } install(WebSockets) { contentConverter = KotlinxWebsocketSerializationConverter(Json) } } val webSocketDataSourceStream: Flow<WSPacket> = receiveMessages() .onStart { openWebSocketSession() sendMessage(WSPacket(id = WsMessage.WsInit, data = "")) } private suspend fun openWebSocketSession() { session = client.webSocketSession( method = HttpMethod.Get, host = Environment.HOST, path = Environment.PATH, port = 8080 ) {
53. private suspend fun openWebSocketSession() { session = client.webSocketSession( method = HttpMethod.Get, host = Environment.HOST, path = Environment.PATH, port = 8080 ) { url.protocol = URLProtocol.WS } } private suspend fun closeWebSocketSession() { session ?. close() } fun close() { client.close() session = null } suspend fun sendMessage(wsPacket: WSPacket) { logger.d { "Sending: $wsPacket" } try { session ?. ensureActive() if (session == null) { logger.e { "Unable to send message since session is null. $wsPacket" } } else { ?.
54. suspend fun sendMessage(wsPacket: WSPacket) { logger.d { "Sending: $wsPacket" } try { session ?. ensureActive() if (session == null) { logger.e { "Unable to send message since session is null. $wsPacket" } } else { session ?. sendSerialized(wsPacket) } } catch (e: Exception) { logger.e(throwable = e.cause, message = { e.message.orEmpty() }) } } private fun receiveMessages(): Flow<WSPacket> { return flow { emit(session !! ) } // This is needed because session is not initialized yet. .flatMapLatest { session -> session.incoming.consumeAsFlow() .map { session.converter !! .deserialize<WSPacket>(it) } .onEach { logger.d { "Receiving: $it" } } } .timeout(TIMEOUT_DURATION) .catch { logger.i { "Timeout exception was triggered" } } .onCompletion { throw CancellationException("WebSocket connection terminated") } } }
55. ConnectionRepository WebSocketDataSource WebRTCDataSource ViewModel
56. WebRTCDataSource
57. class WebRTCDataSource( private val peerConnection: PeerConnection = PeerConnection(defaultRtcConfig), private var dataChannel: DataChannel? = null ) { private var peerConnectionJob: Job = SupervisorJob() private val coroutineScope: CoroutineScope = CoroutineScope(Dispatchers.Default) var onIceCandidateStream: Flow<IceCandidate>? = null var peerConnectionState: StateFlow<PeerConnectionState>? = null init { dataChannel = peerConnection.createDataChannel( label = "input", id = 0, ordered = true, maxRetransmitTimeMs = -1, maxRetransmits = -1, protocol = "", negotiated = false ) onIceCandidateStream = peerConnection.onIceCandidate peerConnectionState = peerConnection.onConnectionStateChange
58. onIceCandidateStream = peerConnection.onIceCandidate peerConnectionState = peerConnection.onConnectionStateChange .stateIn( scope = coroutineScope + peerConnectionJob, started = SharingStarted.WhileSubscribed(), initialValue = PeerConnectionState.New ) } val onMessageDataChannelStream = dataChannel ?. onMessage ?. shareIn( scope = coroutineScope + peerConnectionJob, started = SharingStarted.WhileSubscribed() ) fun sendCommand(key: NavKey, state: NavState) { logger.d { "sendCommand: ${key.name} ${state.name}" } val code = key.ordinal + state.ordinal.times(128) dataChannel ?. send(byteArrayOf(code.toByte())) } suspend fun createOffer(): SessionDescription = peerConnection.createOffer(defaultOfferAnswerOptions) .also { peerConnection.setLocalDescription(it) }
59. suspend fun createOffer(): SessionDescription = peerConnection.createOffer(defaultOfferAnswerOptions) .also { peerConnection.setLocalDescription(it) } suspend fun createAnswer(wsSDPAnswerData: WsSDPAnswerData) = SessionDescription( type = SessionDescriptionType.Answer, sdp = wsSDPAnswerData.sdp ).also { peerConnection.setRemoteDescription(it) } fun createRtcConfiguration(iceServers: List<WsIceServer>) = RtcConfiguration( iceServers = iceServers.map { iceServer -> IceServer( urls = iceServer.urls, password = iceServer.credential.orEmpty(), username = iceServer.username.orEmpty() ) } ) fun setConfiguration(configuration: RtcConfiguration) { peerConnection.setConfiguration(configuration) } fun addIceCandidates(iceCandidates: List<WsIceCandidateData>) { iceCandidates.forEach { wsIceCandidate -> peerConnection.addIceCandidate( IceCandidate(
60. sdpMid = wsIceCandidate.sdpMid, sdpMLineIndex = wsIceCandidate.sdpMLineIndex, candidate = wsIceCandidate.candidate, ) ) } } fun closeWebRTCSession() { logger.i { "closeWebRTCConnection" } peerConnectionJob.cancel() dataChannel = null peerConnection.close() } companion object { val defaultRtcConfig = RtcConfiguration( iceServers = listOf( IceServer( urls = listOf( "stun:stun1.l.google.com:19302", "stun:stun2.l.google.com:19302" ) ), ) )
61. ConnectionRepository WebSocketDataSource WebRTCDataSource ViewModel
62. ConnectionRepository
63. class DefaultConnectionRepository( private val coroutineScope: CoroutineScope, private val webSocketDataSource: WebSocketDataSource, private val webRTCDataSource: WebRTCDataSource, var connectionInfo: ConnectionInfo, ) : ConnectionRepository { override val controllerState: StateFlow<ControllerState> = combine( webRTCDataSource.peerConnectionState, webSocketDataSource.webSocketDataSourceStream, ) { peerConnectionState, packet -> when (packet.id) { WsMessage.WsError -> { val wsErrorCodeData: WsErrorCodeData = Json.decodeFromString(packet.data) PairingState.Failed(message = wsErrorCodeData.message) } else -> { when (peerConnectionState) { PeerConnectionState.New -> {
64. when (packet.id) { WsMessage.WsIceServers -> { val wsIceServersData = Json.decodeFromString<WsIceServersData>(packet.data) val iceServers = wsIceServersData.iceServers val rtcConfigurations = webRTCDataSource.createRtcConfiguration(iceServers) webRTCDataSource.setConfiguration(rtcConfigurations) sendWsIceCandidate() val sessionDescription = webRTCDataSource.createOffer() webSocketDataSource.sendMessage( WSPacket( id = WsMessage.WsSDPOffer, data = Json.encodeToString( WsSDPOfferData( type = sessionDescription.type.name.lowercase(), sdp = sessionDescription.sdp, controllerId = connectionInfo.controllerId, )
65. PairingState.Connecting } WsMessage.WsIceCandidates -> { val wsIceCandidateDataList = Json.decodeFromString<WsIceCandidatesData>(packet.data) webRTCDataSource.addIceCandidates(wsIceCandidateDataList.iceCandidates) PairingState.Connecting } WsMessage.WsSDPAnswer -> { val wsSDPAnswerData = Json.decodeFromString<WsSDPAnswerData>(packet.data) webRTCDataSource.createAnswer(wsSDPAnswerData) PairingState.Connecting } else -> PairingState.Connecting } } PeerConnectionState.Connecting -> { when (packet.id) { WsMessage.WsSDPAnswer -> {
66. PeerConnectionState.Connecting -> { when (packet.id) { WsMessage.WsSDPAnswer -> { val wsSDPAnswerData = Json.decodeFromString<WsSDPAnswerData>(packet.data) webRTCDataSource.createAnswer(wsSDPAnswerData) PairingState.Connecting } WsMessage.WsIceCandidates -> { val wsIceCandidateDataList = Json.decodeFromString<WsIceCandidatesData>(packet.data) webRTCDataSource.addIceCandidates(wsIceCandidateDataList.iceCandidates) PairingState.Connecting } else -> PairingState.Connecting } } PeerConnectionState.Connected -> PairingState.Connected PeerConnectionState.Disconnected -> PairingState.Disconnected PeerConnectionState.Failed -> PairingState.Failed( message = "PeerConnectionState.Failed"
67. PeerConnectionState.Connected -> PairingState.Connected PeerConnectionState.Disconnected -> PairingState.Disconnected PeerConnectionState.Failed -> PairingState.Failed( message = "PeerConnectionState.Failed" ) PeerConnectionState.Closed -> PairingState.Closed } } } } .catch { emit(PairingState.Disconnected) } .onCompletion { webRTCDataSource.closeWebRTCSession() webSocketDataSource.close() } .stateIn( scope = coroutineScope, started = SharingStarted.WhileSubscribed(), initialValue = PairingState.New ) override suspend fun startSessions(connectionInfo: ConnectionInfo) {
68. private fun sendWsIceCandidate() { coroutineScope.launch { webRTCDataSource.onIceCandidateStream ?. collect { iceCandidate -> webSocketDataSource.sendMessage( WSPacket( id = WsMessage.WsIceCandidates, data = Json.encodeToString( WsIceCandidatesData( controllerId = connectionInfo.controllerId, iceCandidates = listOf( WsIceCandidateData( candidate = iceCandidate.candidate, sdpMLineIndex = iceCandidate.sdpMLineIndex, sdpMid = iceCandidate.sdpMid, ) ) ) ) ) ) } } } }
69. 🧑💻 Multiplatform
72. @Composable expect fun Video( videoTrack: VideoStreamTrack, modifier: Modifier = Modifier, audioTrack: AudioStreamTrack? = null ) https: // github.com/shepeliev/webrtc-kmp
73. AndroidView( modifier = modifier, factory = { context -> SurfaceViewRenderer(context).apply { setScalingType( RendererCommon.ScalingType.SCALE_ASPECT_BALANCED, RendererCommon.ScalingType.SCALE_ASPECT_FIT ) renderer = this } }, ) https: // github.com/shepeliev/webrtc-kmp
74. @OptIn(ExperimentalForeignApi :: class) @Composable actual fun Video( videoTrack: VideoStreamTrack, modifier: Modifier, audioTrack: AudioStreamTrack?) { UIKitView( factory = { RTCMTLVideoView().apply { contentMode = UIViewContentMode.UIViewContentModeScaleAspectFit videoTrack.addRenderer(this) } }, modifier = modifier, ) } https: // github.com/shepeliev/webrtc-kmp
75. ▶︎
76. 🤔 What is next?
77. 👨✈ Selective Forwarding Unit Client A Client B
78. 👨✈ Selective Forwarding Unit Client A Client B Client C
79. 👨✈ Selective Forwarding Unit Client A Client B Client C Server PeerA → SFU → PeerB |→ PeerC PeerB → SFU → PeerA |→ PeerC PeerC → SFU → PeerA |→ PeerB
80. 👨✈ Selective Forwarding Unit Server Client A Client B Client C Client D
81. 👨✈ Selective Forwarding Unit Common technologies used with an SFU • Simulcast • Temporal scalability • Insertable Streams • Protocol Converter
82. What is next? Why WebRTC Companies Exist? • Scalability and Performance • Security • Customization • Easy to setup
83. What is next? Why WebRTC Companies Exist? • GetStream • LiveKit • Dyte • Zoom SDK • Agora • Vonage • Twilio Video • Mux • Daily • AWS Chime • Whereby • MirrorFly
84. What is next?
85. Thank you! X: @renaud_mathieu GitHub: @renaudmathieu