Skip to content

Commit 2eed7b1

Browse files
zibet27zibet27
committed
Introduce WebRTC IO and DataChannelClosed exceptions; add data channel iterator
1 parent 2be67f4 commit 2eed7b1

10 files changed

Lines changed: 206 additions & 25 deletions

File tree

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

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -52,22 +52,25 @@ public class AndroidWebRtcDataChannel(
5252
override val protocol: String
5353
get() = channelInit?.protocol ?: error("Protocol is not supported in WebRTC")
5454

55-
private fun assertOpen() {
56-
if (!state.canSend()) {
57-
error("Data channel is closed.")
58-
}
55+
private fun requireOpen() {
56+
if (state.canSend()) return
57+
throw WebRtc.DataChannelClosedException("Data channel '$label' cannot send.")
5958
}
6059

6160
override suspend fun send(text: String) {
62-
assertOpen()
61+
requireOpen()
6362
val buffer = DataChannel.Buffer(Charsets.UTF_8.encode(text), false)
64-
nativeChannel.send(buffer)
63+
if (!nativeChannel.send(buffer)) {
64+
throw WebRtc.IOException("Failed to send text message over data channel '$label'.")
65+
}
6566
}
6667

6768
override suspend fun send(bytes: ByteArray) {
68-
assertOpen()
69+
requireOpen()
6970
val buffer = DataChannel.Buffer(ByteBuffer.wrap(bytes), true)
70-
nativeChannel.send(buffer)
71+
if (!nativeChannel.send(buffer)) {
72+
throw WebRtc.IOException("Failed to send binary message over data channel '$label'.")
73+
}
7174
}
7275

7376
override fun setBufferedAmountLowThreshold(threshold: Long) {

‎ktor-client/ktor-client-webrtc/api/jvm/ktor-client-webrtc.api‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,11 @@ public final class io/ktor/client/webrtc/WebRtc$DataChannel$State : java/lang/En
174174
public static fun values ()[Lio/ktor/client/webrtc/WebRtc$DataChannel$State;
175175
}
176176

177+
public class io/ktor/client/webrtc/WebRtc$DataChannelClosedException : io/ktor/client/webrtc/WebRtc$IOException {
178+
public fun <init> (Ljava/lang/String;Ljava/lang/Throwable;)V
179+
public synthetic fun <init> (Ljava/lang/String;Ljava/lang/Throwable;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
180+
}
181+
177182
public final class io/ktor/client/webrtc/WebRtc$DegradationPreference : java/lang/Enum {
178183
public static final field BALANCED Lio/ktor/client/webrtc/WebRtc$DegradationPreference;
179184
public static final field DISABLED Lio/ktor/client/webrtc/WebRtc$DegradationPreference;
@@ -190,6 +195,11 @@ public abstract interface class io/ktor/client/webrtc/WebRtc$DtmfSender {
190195
public abstract fun insertDtmf (Ljava/lang/String;II)V
191196
}
192197

198+
public class io/ktor/client/webrtc/WebRtc$IOException : java/io/IOException {
199+
public fun <init> (Ljava/lang/String;Ljava/lang/Throwable;)V
200+
public synthetic fun <init> (Ljava/lang/String;Ljava/lang/Throwable;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
201+
}
202+
193203
public final class io/ktor/client/webrtc/WebRtc$IceCandidate {
194204
public static final field Companion Lio/ktor/client/webrtc/WebRtc$IceCandidate$Companion;
195205
public fun <init> (Ljava/lang/String;Ljava/lang/String;I)V
@@ -487,6 +497,7 @@ public abstract class io/ktor/client/webrtc/WebRtcDataChannel : io/ktor/client/w
487497
public fun <init> (Lio/ktor/client/webrtc/DataChannelReceiveOptions;)V
488498
public fun close ()V
489499
protected final fun emitMessage-JP2dKIU (Lio/ktor/client/webrtc/WebRtc$DataChannel$Message;)Ljava/lang/Object;
500+
public final fun iterator ()Lkotlinx/coroutines/channels/ChannelIterator;
490501
public fun receive (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
491502
public fun receiveBinary (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
492503
public fun receiveText (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
@@ -496,6 +507,10 @@ public abstract class io/ktor/client/webrtc/WebRtcDataChannel : io/ktor/client/w
496507
public fun tryReceiveText ()Ljava/lang/String;
497508
}
498509

510+
public final class io/ktor/client/webrtc/WebRtcDataChannelKt {
511+
public static final fun withIOException (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
512+
}
513+
499514
public final class io/ktor/client/webrtc/WebRtcDataChannelOptions {
500515
public fun <init> ()V
501516
public final fun getId ()Ljava/lang/Integer;

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ abstract class io.ktor.client.webrtc/WebRtcDataChannel : io.ktor.client.webrtc/W
112112
constructor <init>(io.ktor.client.webrtc/DataChannelReceiveOptions) // io.ktor.client.webrtc/WebRtcDataChannel.<init>|<init>(io.ktor.client.webrtc.DataChannelReceiveOptions){}[0]
113113

114114
final fun emitMessage(io.ktor.client.webrtc/WebRtc.DataChannel.Message): kotlinx.coroutines.channels/ChannelResult<kotlin/Unit> // io.ktor.client.webrtc/WebRtcDataChannel.emitMessage|emitMessage(io.ktor.client.webrtc.WebRtc.DataChannel.Message){}[0]
115+
final fun iterator(): kotlinx.coroutines.channels/ChannelIterator<io.ktor.client.webrtc/WebRtc.DataChannel.Message> // io.ktor.client.webrtc/WebRtcDataChannel.iterator|iterator(){}[0]
115116
final fun stopReceivingMessages() // io.ktor.client.webrtc/WebRtcDataChannel.stopReceivingMessages|stopReceivingMessages(){}[0]
116117
open fun tryReceive(): io.ktor.client.webrtc/WebRtc.DataChannel.Message? // io.ktor.client.webrtc/WebRtcDataChannel.tryReceive|tryReceive(){}[0]
117118
open fun tryReceiveBinary(): kotlin/ByteArray? // io.ktor.client.webrtc/WebRtcDataChannel.tryReceiveBinary|tryReceiveBinary(){}[0]
@@ -689,6 +690,14 @@ final object io.ktor.client.webrtc/WebRtc { // io.ktor.client.webrtc/WebRtc|null
689690
final fun hashCode(): kotlin/Int // io.ktor.client.webrtc/WebRtc.Stats.hashCode|hashCode(){}[0]
690691
final fun toString(): kotlin/String // io.ktor.client.webrtc/WebRtc.Stats.toString|toString(){}[0]
691692
}
693+
694+
open class DataChannelClosedException : io.ktor.client.webrtc/WebRtc.IOException { // io.ktor.client.webrtc/WebRtc.DataChannelClosedException|null[0]
695+
constructor <init>(kotlin/String, kotlin/Throwable? = ...) // io.ktor.client.webrtc/WebRtc.DataChannelClosedException.<init>|<init>(kotlin.String;kotlin.Throwable?){}[0]
696+
}
697+
698+
open class IOException : kotlinx.io/IOException { // io.ktor.client.webrtc/WebRtc.IOException|null[0]
699+
constructor <init>(kotlin/String, kotlin/Throwable? = ...) // io.ktor.client.webrtc/WebRtc.IOException.<init>|<init>(kotlin.String;kotlin.Throwable?){}[0]
700+
}
692701
}
693702

694703
final object io.ktor.client.webrtc/WebRtcMedia { // io.ktor.client.webrtc/WebRtcMedia|null[0]
@@ -828,6 +837,7 @@ final object io.ktor.client.webrtc/WebRtcMedia { // io.ktor.client.webrtc/WebRtc
828837
}
829838

830839
final fun <#A: io.ktor.client.webrtc/WebRtcConfig> io.ktor.client.webrtc/WebRtcClient(io.ktor.client.webrtc/WebRtcClientEngineFactory<#A>, kotlin/Function1<#A, kotlin/Unit> = ...): io.ktor.client.webrtc/WebRtcClient // io.ktor.client.webrtc/WebRtcClient|WebRtcClient(io.ktor.client.webrtc.WebRtcClientEngineFactory<0:0>;kotlin.Function1<0:0,kotlin.Unit>){0§<io.ktor.client.webrtc.WebRtcConfig>}[0]
840+
final suspend inline fun <#A: kotlin/Any?> io.ktor.client.webrtc/withIOException(crossinline kotlin.coroutines/SuspendFunction0<#A>): #A // io.ktor.client.webrtc/withIOException|withIOException(kotlin.coroutines.SuspendFunction0<0:0>){0§<kotlin.Any?>}[0]
831841

832842
// Targets: [ios]
833843
abstract interface io.ktor.client.webrtc.media/Capturer : kotlin/AutoCloseable { // io.ktor.client.webrtc.media/Capturer|null[0]

‎ktor-client/ktor-client-webrtc/common/src/io/ktor/client/webrtc/WebRtc.kt‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -476,6 +476,8 @@ public object WebRtc {
476476
* [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.webrtc.WebRtc.DataChannel.send)
477477
*
478478
* @param text The text message to send.
479+
* @throws DataChannelClosedException if the channel is closed or not open for sending.
480+
* @throws IOException if the message cannot be sent by the underlying WebRTC transport.
479481
* @see [MDN RTCDataChannel.send()](https://developer.mozilla.org/en-US/docs/Web/API/RTCDataChannel/send)
480482
*/
481483
public suspend fun send(text: String)
@@ -486,6 +488,8 @@ public object WebRtc {
486488
* [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.webrtc.WebRtc.DataChannel.send)
487489
*
488490
* @param bytes The binary data to send.
491+
* @throws DataChannelClosedException if the channel is closed or not open for sending.
492+
* @throws IOException if the message cannot be sent by the underlying WebRTC transport.
489493
* @see [MDN RTCDataChannel.send()](https://developer.mozilla.org/en-US/docs/Web/API/RTCDataChannel/send)
490494
*/
491495
public suspend fun send(bytes: ByteArray)
@@ -496,6 +500,8 @@ public object WebRtc {
496500
* This method will suspend the current coroutine until a message is received.
497501
* The message can be either text or binary data.
498502
*
503+
* @throws DataChannelClosedException if the channel is closed by this or remote peer.
504+
*
499505
* [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.webrtc.WebRtc.DataChannel.receive)
500506
*/
501507
public suspend fun receive(): Message
@@ -593,4 +599,21 @@ public object WebRtc {
593599
* [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.webrtc.WebRtc.IceException)
594600
*/
595601
public class IceException(message: String?, cause: Throwable? = null) : RuntimeException(message, cause)
602+
603+
/**
604+
* Signals that some I/O exception has occurred.
605+
*
606+
* [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.webrtc.WebRtc.IOException)
607+
*/
608+
public open class IOException(message: String, cause: Throwable? = null) : kotlinx.io.IOException(message, cause)
609+
610+
/**
611+
* Exception thrown when trying to send to or read from a closed [DataChannel].
612+
*
613+
* [Report a problem](https://ktor.io/feedback/?fqname=io.ktor.client.webrtc.WebRtc.DataChannelClosedException)
614+
*/
615+
public open class DataChannelClosedException(
616+
message: String,
617+
cause: Throwable? = null
618+
) : IOException(message, cause)
596619
}

‎ktor-client/ktor-client-webrtc/common/src/io/ktor/client/webrtc/WebRtcDataChannel.kt‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -205,7 +205,24 @@ public abstract class WebRtcDataChannel private constructor(
205205
receiveChannel = Channel(options = receiveOptions)
206206
)
207207

208-
override suspend fun receive(): WebRtc.DataChannel.Message = receiveChannel.receive()
208+
/**
209+
* Returns an iterator over received messages.
210+
*
211+
* If the data channel is closed, the iterator keeps returning messages already stored in the buffer
212+
* and completes when no buffered messages remain.
213+
*/
214+
public operator fun iterator(): ChannelIterator<WebRtc.DataChannel.Message> {
215+
return receiveChannel.iterator()
216+
}
217+
218+
override suspend fun receive(): WebRtc.DataChannel.Message {
219+
val result = receiveChannel.receiveCatching()
220+
if (result.isSuccess) {
221+
return result.getOrThrow()
222+
}
223+
val message = "Data channel '$label' is closed and no more messages will be received."
224+
throw WebRtc.DataChannelClosedException(message, result.exceptionOrNull())
225+
}
209226

210227
override suspend fun receiveBinary(): ByteArray = receive().binaryOrThrow()
211228

@@ -225,3 +242,16 @@ public abstract class WebRtcDataChannel private constructor(
225242
receiveChannel.close()
226243
}
227244
}
245+
246+
@InternalAPI
247+
public suspend inline fun <R> withIOException(crossinline block: suspend () -> R): R {
248+
return try {
249+
block()
250+
} catch (cause: kotlinx.coroutines.CancellationException) {
251+
throw cause
252+
} catch (cause: WebRtc.IOException) {
253+
throw cause
254+
} catch (cause: Exception) {
255+
throw WebRtc.IOException("Error in WebRtcDataChannel operation", cause)
256+
}
257+
}

‎ktor-client/ktor-client-webrtc/common/test/io/ktor/client/webrtc/WebRtcDataChannelIteratorTest.kt‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
/*
2+
* Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
3+
*/
4+
5+
package io.ktor.client.webrtc
6+
7+
import kotlinx.coroutines.test.*
8+
import kotlin.test.*
9+
10+
class WebRtcDataChannelIteratorTest {
11+
12+
@Test
13+
fun testIteratorDrainsBufferedMessagesAfterChannelIsClosed() = runTest {
14+
val channel = TestWebRtcDataChannel()
15+
channel.emit(WebRtc.DataChannel.Message.Text("first"))
16+
channel.emit(WebRtc.DataChannel.Message.Text("second"))
17+
channel.closeReceiving()
18+
19+
val messages = channel.toList()
20+
21+
assertEquals(listOf("first", "second"), messages.map { it.textOrThrow() })
22+
}
23+
24+
@Test
25+
fun testIteratorStopsOnClosedChannelWithoutBufferedMessages() = runTest {
26+
val channel = TestWebRtcDataChannel()
27+
channel.closeReceiving()
28+
29+
assertTrue(channel.toList().isEmpty())
30+
}
31+
32+
@Test
33+
fun testIteratorPreservesMessageOrderAndType() = runTest {
34+
val channel = TestWebRtcDataChannel()
35+
val bytes = byteArrayOf(1, 2, 3)
36+
channel.emit(WebRtc.DataChannel.Message.Text("text"))
37+
channel.emit(WebRtc.DataChannel.Message.Binary(bytes))
38+
channel.closeReceiving()
39+
40+
val messages = channel.toList()
41+
42+
assertEquals("text", messages[0].textOrThrow())
43+
assertContentEquals(bytes, messages[1].binaryOrThrow())
44+
}
45+
46+
private suspend fun WebRtcDataChannel.toList(): List<WebRtc.DataChannel.Message> = buildList {
47+
for (message in this@toList) add(message)
48+
}
49+
50+
private class TestWebRtcDataChannel : WebRtcDataChannel(DataChannelReceiveOptions()) {
51+
override val id: Int? = null
52+
override val label: String = "test"
53+
override var state: WebRtc.DataChannel.State = WebRtc.DataChannel.State.OPEN
54+
override val bufferedAmount: Long = 0
55+
override val bufferedAmountLowThreshold: Long = 0
56+
override val maxPacketLifeTime: Int? = null
57+
override val maxRetransmits: Int? = null
58+
override val negotiated: Boolean = false
59+
override val ordered: Boolean = true
60+
override val protocol: String = ""
61+
62+
fun emit(message: WebRtc.DataChannel.Message) {
63+
check(emitMessage(message).isSuccess)
64+
}
65+
66+
override fun setBufferedAmountLowThreshold(threshold: Long) = TODO()
67+
68+
override suspend fun send(text: String) = TODO()
69+
70+
override suspend fun send(bytes: ByteArray) = TODO()
71+
72+
fun closeReceiving() {
73+
state = WebRtc.DataChannel.State.CLOSED
74+
stopReceivingMessages()
75+
}
76+
77+
override fun closeTransport() {
78+
closeReceiving()
79+
}
80+
}
81+
}

‎ktor-client/ktor-client-webrtc/common/test/io/ktor/client/webrtc/WebRtcDataChannelTest.kt‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -209,9 +209,9 @@ class WebRtcDataChannelTest {
209209
dataChannel1.waitForClose(dataChannelEvents1)
210210
dataChannel2.waitForClose(dataChannelEvents2)
211211

212-
assertFails { dataChannel1.send("Hello") }
213-
assertFails { dataChannel2.send("Hello") }
214-
assertFails { dataChannel1.receive() }
212+
assertFailsWith<WebRtc.DataChannelClosedException> { dataChannel1.send("Hello") }
213+
assertFailsWith<WebRtc.DataChannelClosedException> { dataChannel2.send("Hello") }
214+
assertFailsWith<WebRtc.DataChannelClosedException> { dataChannel1.receive() }
215215
assertEquals(null, dataChannel2.tryReceive())
216216
}
217217

‎ktor-client/ktor-client-webrtc/ios/src/io/ktor/client/webrtc/DataChannel.kt‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -59,23 +59,22 @@ public class IosWebRtcDataChannel(
5959
override val protocol: String
6060
get() = nativeChannel.protocol()
6161

62-
private fun assertOpen() {
63-
if (!state.canSend()) {
64-
error("Data channel is closed.")
65-
}
62+
private fun requireOpen() {
63+
if (state.canSend()) return
64+
throw WebRtc.DataChannelClosedException("Data channel '$label' cannot send.")
6665
}
6766

6867
override suspend fun send(text: String) {
69-
assertOpen()
68+
requireOpen()
7069
if (!nativeChannel.sendData(data = text.toRTCDataBuffer())) {
71-
error("Failed to send text message over DataChannel.")
70+
throw WebRtc.IOException("Failed to send text message over data channel '$label'.")
7271
}
7372
}
7473

7574
override suspend fun send(bytes: ByteArray) {
76-
assertOpen()
75+
requireOpen()
7776
if (!nativeChannel.sendData(data = bytes.toRTCDataBuffer())) {
78-
error("Failed to send binary message over DataChannel.")
77+
throw WebRtc.IOException("Failed to send binary message over data channel '$label'.")
7978
}
8079
}
8180

‎ktor-client/ktor-client-webrtc/ktor-client-webrtc-rs/common/src/io/ktor/client/webrtc/rs/DataChannel.kt‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
/*
2-
* Copyright 2014-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
2+
* Copyright 2014-2026 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
33
*/
44

55
package io.ktor.client.webrtc.rs
66

77
import io.ktor.client.webrtc.*
8+
import io.ktor.utils.io.InternalAPI
89
import kotlinx.coroutines.CoroutineScope
910
import kotlinx.coroutines.CoroutineStart
1011
import kotlinx.coroutines.launch
@@ -55,12 +56,21 @@ public class RustWebRtcDataChannel(
5556
override val protocol: String
5657
get() = inner.protocol()
5758

59+
private fun requireOpen() {
60+
if (state.canSend()) return
61+
throw WebRtc.DataChannelClosedException("Data channel '$label' cannot send.")
62+
}
63+
64+
@OptIn(InternalAPI::class)
5865
override suspend fun send(text: String) {
59-
inner.sendText(text)
66+
requireOpen()
67+
withIOException { inner.sendText(text) }
6068
}
6169

70+
@OptIn(InternalAPI::class)
6271
override suspend fun send(bytes: ByteArray) {
63-
inner.send(bytes)
72+
requireOpen()
73+
withIOException { inner.send(bytes) }
6474
}
6575

6676
override fun setBufferedAmountLowThreshold(threshold: Long): Unit = runBlocking {

‎ktor-client/ktor-client-webrtc/web/src/io/ktor/client/webrtc/DataChannel.kt‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
package io.ktor.client.webrtc
66

7+
import io.ktor.utils.io.InternalAPI
78
import js.buffer.ArrayBuffer
89
import js.buffer.toByteArray
910
import js.typedarrays.toInt8Array
@@ -60,12 +61,21 @@ public class JsWebRtcDataChannel(
6061
override val protocol: String
6162
get() = channel.protocol
6263

64+
private fun requireOpen() {
65+
if (state.canSend()) return
66+
throw WebRtc.DataChannelClosedException("Data channel '$label' cannot send.")
67+
}
68+
69+
@OptIn(InternalAPI::class)
6370
override suspend fun send(text: String) {
64-
channel.send(text)
71+
requireOpen()
72+
withIOException { channel.send(text) }
6573
}
6674

75+
@OptIn(InternalAPI::class)
6776
override suspend fun send(bytes: ByteArray) {
68-
channel.send(bytes.toInt8Array())
77+
requireOpen()
78+
withIOException { channel.send(bytes.toInt8Array()) }
6979
}
7080

7181
override fun setBufferedAmountLowThreshold(threshold: Long) {

0 commit comments

Comments
 (0)