From b908347b1e6c1dd8e044b5a577f12fed4462d816 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 01:04:06 +0200 Subject: [PATCH 1/3] fix: check the byte array and catch C++ exceptions in the synchronous data channel sends sendByteArrayBuffer used the result of GetByteArrayElements unchecked and built the buffer outside the catch; both sync paths now follow the async ones. --- .../src/main/cpp/src/JNI_RTCDataChannel.cpp | 47 +++++++++++-------- 1 file changed, 28 insertions(+), 19 deletions(-) diff --git a/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp b/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp index 2cdcea0d..395a02fa 100644 --- a/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp +++ b/webrtc-jni/src/main/cpp/src/JNI_RTCDataChannel.cpp @@ -201,39 +201,48 @@ JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_dispose JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendDirectBuffer (JNIEnv * env, jobject caller, jobject jBuffer, jboolean isBinary) { - webrtc::DataChannelInterface * channel = GetHandle(env, caller); - CHECK_HANDLE(channel); + try { + webrtc::DataChannelInterface * channel = GetHandle(env, caller); + CHECK_HANDLE(channel); - uint8_t * address = static_cast(env->GetDirectBufferAddress(jBuffer)); + uint8_t * address = static_cast(env->GetDirectBufferAddress(jBuffer)); - if (address != NULL) { - jlong bufferLength = env->GetDirectBufferCapacity(jBuffer); + if (address != NULL) { + jlong bufferLength = env->GetDirectBufferCapacity(jBuffer); - webrtc::CopyOnWriteBuffer data(address, static_cast(bufferLength)); + webrtc::CopyOnWriteBuffer data(address, static_cast(bufferLength)); - if (!channel->Send(webrtc::DataBuffer(data, static_cast(isBinary)))) { - env->Throw(jni::JavaRuntimeException(env, "Data channel rejected the send")); + if (!channel->Send(webrtc::DataBuffer(data, static_cast(isBinary)))) { + env->Throw(jni::JavaRuntimeException(env, "Data channel rejected the send")); + } + } + else { + env->Throw(jni::JavaError(env, "Non-direct buffer provided")); } } - else { - env->Throw(jni::JavaError(env, "Non-direct buffer provided")); + catch (...) { + ThrowCxxJavaException(env); } } JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_RTCDataChannel_sendByteArrayBuffer (JNIEnv * env, jobject caller, jbyteArray jBufferArray, jboolean isBinary) { - webrtc::DataChannelInterface * channel = GetHandle(env, caller); - CHECK_HANDLE(channel); - - int8_t * arrayPtr = env->GetByteArrayElements(jBufferArray, nullptr); - size_t arrayLength = env->GetArrayLength(jBufferArray); - - webrtc::CopyOnWriteBuffer data(arrayPtr, arrayLength); + try { + webrtc::DataChannelInterface * channel = GetHandle(env, caller); + CHECK_HANDLE(channel); - env->ReleaseByteArrayElements(jBufferArray, arrayPtr, JNI_ABORT); + auto releaseArray = [env, jBufferArray](jbyte * bytes) { + env->ReleaseByteArrayElements(jBufferArray, bytes, JNI_ABORT); + }; + std::unique_ptr bytes( + env->GetByteArrayElements(jBufferArray, nullptr), releaseArray); + if (!bytes) { + return; + } + webrtc::CopyOnWriteBuffer data(bytes.get(), env->GetArrayLength(jBufferArray)); + bytes.reset(); - try { if (!channel->Send(webrtc::DataBuffer(data, static_cast(isBinary)))) { env->Throw(jni::JavaRuntimeException(env, "Data channel rejected the send")); } From a5b77b8ad9f62b54dada6103ff8b73471d60f6d4 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 01:04:06 +0200 Subject: [PATCH 2/3] fix: fail WebRtc.closeAsync when the factory is left undisposed A close that leaks the factory to protect WebRTC's threads no longer completes normally, so callers can tell it from a full release. --- docs/guide/client-api.md | 2 + .../java/dev/onvoid/webrtc/client/WebRtc.java | 60 ++++++++++++++----- .../dev/onvoid/webrtc/client/PeerTests.java | 38 ++++++++++++ 3 files changed, 84 insertions(+), 16 deletions(-) diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 26995757..bbd2c150 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -613,6 +613,8 @@ You close what you created, and closing an owner closes what it owns: `WebRtc.close()` is the one close that waits, for up to the close timeout, which makes it fit for try-with-resources. Once it returns, every peer's `state()` is `CLOSED`; the `whenClosed` futures and `CLOSED` events follow on the peers' event threads, since waiting for them could never end when `close()` is called on an event thread. +When a peer or track does not close within the close timeout, the runtime leaves its factory undisposed rather than free what WebRTC's threads may still use. `WebRtc.closeAsync()` then fails with a `WebRtcException` naming the cause, and the error handler hears of it as well, which is how `close()` reports it. + ## Errors Failures arrive as exceptions, thrown or as the cause of a failed future, all of them a `WebRtcException`: diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java index 296d2d13..2bfd642b 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -26,12 +26,14 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; @@ -280,8 +282,14 @@ public AudioMode audioMode() { /** * Closes every peer, then every track, then the factory, and stops the * threads of this API. + *

+ * When a peer or track does not close within the close timeout, the + * factory is left undisposed, since disposing it could free what WebRTC's + * threads still use. The future then fails with a {@link WebRtcException} + * that names the cause, as it does when disposing fails. * - * @return a future that completes once all is closed. + * @return a future that completes once all is closed, or fails when not + * everything could be released. */ public CompletableFuture closeAsync() { if (closing.compareAndSet(false, true)) { @@ -313,7 +321,10 @@ public void close() { catch (InterruptedException e) { Thread.currentThread().interrupt(); } - catch (Exception e) { + catch (ExecutionException e) { + // Closing reported its failure already. + } + catch (TimeoutException e) { reportError(new WebRtcException("The runtime did not close within " + closeTimeout, e)); } } @@ -350,23 +361,30 @@ CompletableFuture whenShutDown() { } private void shutdown() { + Throwable failure = null; + try { // Peers and tracks created while the runtime closes are closed as // well, so each is waited for until none is left. - boolean released = awaitAll(() -> { + failure = awaitAll(() -> { List> closing = new ArrayList<>(); for (PeerImpl peer : new ArrayList<>(peers)) { closing.add(peer.closeInternal()); } return closing; - }, "peers") && awaitAll(media::closeAll, "local tracks"); + }, "peers"); + + if (failure == null) { + failure = awaitAll(media::closeAll, "local tracks"); + } - if (!released) { + if (failure != null) { // Something may still run on WebRTC's threads, and disposing // the factory would free them under it. Leaking is the lesser harm. - reportError(new WebRtcException("The factory of the runtime was left undisposed, " - + "since not everything it created closed within " + closeTimeout)); + failure = new WebRtcException("The factory of the runtime was left undisposed, " + + "since not everything it created closed", failure); + reportError(failure); return; } @@ -389,25 +407,32 @@ private void shutdown() { } } catch (Throwable e) { + failure = e; reportError(e); } finally { shutdownPools(); - closed.complete(null); + + if (failure == null) { + closed.complete(null); + } + else { + closed.completeExceptionally(failure); + } } } /** * Closes what the supplier names, again and again, until it names nothing. * - * @return false if a round did not finish within the close timeout. + * @return null once all is closed, or why a round did not finish. */ - private boolean awaitAll(Supplier>> closes, String what) { + private WebRtcException awaitAll(Supplier>> closes, String what) { for (int round = 0; round < 10; round++) { List> futures = closes.get(); if (futures.isEmpty()) { - return true; + return null; } try { @@ -416,15 +441,18 @@ private boolean awaitAll(Supplier>> closes, String } catch (InterruptedException e) { Thread.currentThread().interrupt(); - return false; + return new WebRtcException("Closing the " + what + " was interrupted", e); + } + catch (TimeoutException e) { + return new WebRtcException("Closing the " + what + " did not finish within " + closeTimeout, e); } - catch (Exception e) { - reportError(new WebRtcException("Closing the " + what + " did not finish within " + closeTimeout, e)); - return false; + catch (ExecutionException e) { + return new WebRtcException("Closing the " + what + " failed", e.getCause()); } } - return closes.get().isEmpty(); + return closes.get().isEmpty() ? null + : new WebRtcException("The " + what + " kept being created while the runtime closed"); } private void shutdownPools() { diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java index 6543025b..7a84cbce 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -29,6 +29,7 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -433,6 +434,43 @@ void createPeerRacingClose() throws Exception { } } + @Test + void closeFailsWhenPeerStaysOpen() throws Exception { + List closeErrors = new CopyOnWriteArrayList<>(); + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(closeErrors::add) + .closeTimeout(Duration.ofMillis(200)) + .build(); + PeerImpl peer = (PeerImpl) runtime.createPeer(manualOptions()); + CountDownLatch release = new CountDownLatch(1); + + // Holds the peer's control executor, so that its teardown cannot run. + peer.control.execute(() -> { + try { + release.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + try { + ExecutionException e = assertThrows(ExecutionException.class, + () -> runtime.closeAsync().get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + assertInstanceOf(WebRtcException.class, e.getCause()); + assertTrue(e.getCause().getMessage().contains("left undisposed"), e.getCause()::getMessage); + + // The error handler hears of it once, also through close(). + runtime.close(); + assertEquals(Collections.singletonList(e.getCause()), closeErrors); + } + finally { + release.countDown(); + } + } + private static PeerOptions manualOptions() { return PeerOptions.builder().negotiation(Negotiation.MANUAL).build(); } From 500f87bb495b6cd844f4c60f7d7a68091678da37 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 01:08:47 +0200 Subject: [PATCH 3/3] feat: bound what a data channel holds waiting to be sent PeerOptions.channelSendLimit (16 MiB by default) caps the bytes each channel holds in Java; a send beyond it fails at once with a ChannelFullException. --- docs/guide/client-api.md | 4 ++ .../webrtc/client/ChannelFullException.java | 37 +++++++++++++++++++ .../dev/onvoid/webrtc/client/DataChannel.java | 8 ++-- .../onvoid/webrtc/client/DataChannelImpl.java | 17 ++++++++- .../dev/onvoid/webrtc/client/PeerImpl.java | 4 ++ .../dev/onvoid/webrtc/client/PeerOptions.java | 25 +++++++++++++ .../onvoid/webrtc/client/ClientTestBase.java | 12 +++++- .../dev/onvoid/webrtc/client/PeerTests.java | 33 ++++++++++++++++- 8 files changed, 132 insertions(+), 8 deletions(-) create mode 100644 webrtc/src/main/java/dev/onvoid/webrtc/client/ChannelFullException.java diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index bbd2c150..732b6ba8 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -102,6 +102,7 @@ Peer peer = rtc.createPeer(PeerOptions.builder() | `negotiation(Negotiation)` | `AUTOMATIC` by default, or `MANUAL`, see [Manual Negotiation](#manual-negotiation). | | `receive(MediaKind...)` | Receives audio or video without sending any, as a viewer does. | | `gatheringTimeout(Duration)` | How long `Gathering.COMPLETE` waits for ICE gathering, 3 seconds by default. See [Manual Negotiation](#manual-negotiation). | +| `channelSendLimit(long)` | How many bytes each data channel may hold waiting to be sent, 16 MiB by default. See [Data Channels](#data-channels). | | `configure(Consumer)` | Anything else the [peer connection configuration](/guide/networking/peer-connection-config) has. | ### Signaling @@ -505,6 +506,8 @@ for (byte[] chunk : chunks) { } ``` +A sender that does not wait still cannot fill Java memory: each channel holds at most 16 MiB waiting to be sent, and a `send` beyond that fails at once with a `ChannelFullException`. `PeerOptions.builder().channelSendLimit(bytes)` changes the limit for every channel of a peer. `bufferedAmount()` tells how much a channel holds. + `ChannelOptions` set how a channel delivers: ```java @@ -623,6 +626,7 @@ Failures arrive as exceptions, thrown or as the cause of a failed future, all of |---|---| | `NegotiationException` | A description or candidate was rejected, or two impolite peers collided. | | `PeerClosedException` | An operation was pending when its peer, channel or runtime closed, or came after that. | +| `ChannelFullException` | A data channel already holds as many bytes waiting to be sent as its peer's `channelSendLimit` allows. | | `MediaException` | A device is missing, busy, refused access, or already open with other options. | | `ConfigurationException` | A call does not fit how the runtime was built, or WebRTC refused a configuration. | | `BlockingCallException` | A future was waited on from an event thread. | diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/ChannelFullException.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/ChannelFullException.java new file mode 100644 index 00000000..ae2febf1 --- /dev/null +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/ChannelFullException.java @@ -0,0 +1,37 @@ +/* + * Copyright 2026 Alex Andres + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.onvoid.webrtc.client; + +/** + * A message was not sent, since its data channel already holds as many bytes + * waiting to be sent as {@link PeerOptions.Builder#channelSendLimit(long)} + * allows. Messages sent later are taken again once the backlog shrank. + */ +public class ChannelFullException extends WebRtcException { + + private static final long serialVersionUID = 1L; + + /** + * Creates the exception with a message that says what went wrong. + * + * @param message the detail message. + */ + public ChannelFullException(String message) { + super(message); + } + +} diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java index 96301b2e..eafd4d6b 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannel.java @@ -54,7 +54,9 @@ public interface DataChannel extends AutoCloseable { * @return a future that completes when the message is handed to the * transport. While a lot of data waits to be sent, the future completes * only once the backlog shrank, so a loop that waits for each send slows - * down by itself. + * down by itself. The future fails at once with a {@link + * ChannelFullException} when the message would take the backlog beyond + * {@link PeerOptions.Builder#channelSendLimit(long)}. */ CompletableFuture send(String text); @@ -64,7 +66,7 @@ public interface DataChannel extends AutoCloseable { * @param data the message. * * @return a future that completes when the message is handed to the - * transport. + * transport, or fails as {@link #send(String)} describes. */ CompletableFuture send(byte[] data); @@ -75,7 +77,7 @@ public interface DataChannel extends AutoCloseable { * @param data the message. * * @return a future that completes when the message is handed to the - * transport. + * transport, or fails as {@link #send(String)} describes. */ CompletableFuture send(ByteBuffer data); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java index 74de5401..fffd4225 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DataChannelImpl.java @@ -37,7 +37,8 @@ *

* The native channel is used on the peer's control executor, where sends wait * while the channel is not open or holds more than {@link #HIGH_WATER_MARK} - * bytes. Handlers run on the peer's event executor. + * bytes. What waits is bounded by the peer's channel send limit, beyond which + * sends fail at once. Handlers run on the peer's event executor. */ final class DataChannelImpl implements DataChannel { @@ -51,6 +52,9 @@ final class DataChannelImpl implements DataChannel { private final String label; + /** The most bytes {@link #held} may reach. */ + private final long sendLimit; + /** Bytes sent and not yet handed to the transport. */ private final AtomicLong held = new AtomicLong(); @@ -93,6 +97,7 @@ final class DataChannelImpl implements DataChannel { DataChannelImpl(PeerImpl peer, String label) { this.peer = peer; this.label = label; + this.sendLimit = peer.channelSendLimit(); this.messageHandlers = new Handlers<>(peer.rtc::reportError); this.stateHandlers = new Handlers<>(peer.rtc::reportError); } @@ -259,8 +264,16 @@ void closeFromPeer(WebRtcException error) { } private CompletableFuture enqueue(byte[] data, boolean binary) { + long size = data.length; + long before = held.getAndUpdate(amount -> amount + size <= sendLimit ? amount + size : amount); + + if (before + size > sendLimit) { + return peer.toPublic(ClientFuture.failed(new ChannelFullException("The data channel " + label + + " holds " + before + " bytes waiting to be sent, and " + size + " more would exceed " + + "its limit of " + sendLimit))); + } + CompletableFuture sent = new CompletableFuture<>(); - held.addAndGet(data.length); peer.control.execute(() -> { if (released || state == ChannelState.CLOSING || state == ChannelState.CLOSED) { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java index e946f8a1..873dc102 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -549,6 +549,10 @@ boolean isClosing() { return closing.get(); } + long channelSendLimit() { + return options.channelSendLimit; + } + /** * Hands a future out to the application: it completes on the event * executor, so its callbacks run there like the handlers. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java index 94c86b32..24765841 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerOptions.java @@ -61,6 +61,8 @@ public final class PeerOptions { final Duration gatheringTimeout; + final long channelSendLimit; + private PeerOptions(Builder builder) { this.name = builder.name; @@ -73,6 +75,7 @@ private PeerOptions(Builder builder) { ? EnumSet.noneOf(MediaKind.class) : EnumSet.copyOf(builder.receive)); this.customizer = builder.customizer; this.gatheringTimeout = builder.gatheringTimeout; + this.channelSendLimit = builder.channelSendLimit; } /** @@ -132,6 +135,8 @@ public static final class Builder { private Duration gatheringTimeout = Duration.ofSeconds(3); + private long channelSendLimit = 16 * 1024 * 1024; + private Builder() { } @@ -254,6 +259,26 @@ public Builder gatheringTimeout(Duration timeout) { return this; } + /** + * Sets how many bytes each data channel of the peer may hold that were + * sent and not yet handed to the transport; 16 MiB by default. A send + * beyond that fails at once with a {@link ChannelFullException} + * instead of letting the backlog grow without bound, and so does a + * single message larger than the limit. + * + * @param bytes the limit, at least 1. + * + * @return this builder. + */ + public Builder channelSendLimit(long bytes) { + if (bytes < 1) { + throw new IllegalArgumentException("The limit must be at least 1 byte, got " + bytes); + } + + this.channelSendLimit = bytes; + return this; + } + /** * Changes anything else of the {@link RTCConfiguration} the peer is * created with, after this builder applied its own values. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/ClientTestBase.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/ClientTestBase.java index a9720af6..42ae6bfb 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/ClientTestBase.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/ClientTestBase.java @@ -30,6 +30,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.function.UnaryOperator; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -81,13 +82,20 @@ void closeRuntime() { * polite. */ Peer[] pair() { + return pair(UnaryOperator.identity()); + } + + /** + * Like {@link #pair()}, with options both peers share. + */ + Peer[] pair(UnaryOperator options) { SignalingPair link = Signaling.loopback(); - Peer caller = rtc.createPeer(PeerOptions.builder() + Peer caller = rtc.createPeer(options.apply(PeerOptions.builder()) .name("caller") .signaling(link.first()) .build()); - Peer callee = rtc.createPeer(PeerOptions.builder() + Peer callee = rtc.createPeer(options.apply(PeerOptions.builder()) .name("callee") .signaling(link.second()) .polite(true) diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java index 7a84cbce..9d7b9a1e 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -98,7 +98,8 @@ void earlyMessagesKept() throws Exception { @Test void earlyMessagesBounded() throws Exception { - Peer[] peers = pair(); + // The sender holds more than the receiver keeps. + Peer[] peers = pair(options -> options.channelSendLimit(32 * 1024 * 1024)); CompletableFuture remote = new CompletableFuture<>(); CompletableFuture dropped = new CompletableFuture<>(); @@ -471,6 +472,36 @@ void closeFailsWhenPeerStaysOpen() throws Exception { } } + @Test + void sendBeyondLimitFails() throws Exception { + // Never negotiated, so the channel stays connecting and holds every send. + Peer peer = rtc.createPeer(PeerOptions.builder() + .negotiation(Negotiation.MANUAL) + .channelSendLimit(1000) + .build()); + DataChannel channel = peer.createChannel("bounded"); + + CompletableFuture first = channel.send(new byte[600]); + ExecutionException full = assertThrows(ExecutionException.class, () -> await(channel.send(new byte[600]))); + CompletableFuture fitting = channel.send(new byte[400]); + + assertInstanceOf(ChannelFullException.class, full.getCause()); + assertEquals(1000, channel.bufferedAmount()); + assertFalse(first.isDone()); + assertFalse(fitting.isDone()); + + // A message larger than the limit never fits. + peer.close(); + Peer other = rtc.createPeer(PeerOptions.builder() + .negotiation(Negotiation.MANUAL) + .channelSendLimit(10) + .build()); + ExecutionException tooLarge = assertThrows(ExecutionException.class, + () -> await(other.createChannel("small").send(new byte[11]))); + + assertInstanceOf(ChannelFullException.class, tooLarge.getCause()); + } + private static PeerOptions manualOptions() { return PeerOptions.builder().negotiation(Negotiation.MANUAL).build(); }