From e052357b311011542091ba3f853a74e29e4caf81 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Thu, 22 Oct 2026 17:26:16 +0200 Subject: [PATCH] test: make the client concurrency tests deterministic The runtime takes a control pool and a timer through package-private builder methods, so tests step through control tasks and fire timeouts themselves. createPeer is split into register and open, the points a closing runtime can meet it at. New tests cover SerialExecutor rejection, createPeer against close at each point, answers that arrive after the session timed out, and how close failures are reported; the HTTP test endpoint holds answers on a latch instead of sleeping. --- .../java/dev/onvoid/webrtc/client/WebRtc.java | 90 +++++-- .../webrtc/client/HttpSessionTests.java | 130 +++++++-- .../onvoid/webrtc/client/ManualExecutor.java | 201 ++++++++++++++ .../dev/onvoid/webrtc/client/ManualTimer.java | 195 ++++++++++++++ .../dev/onvoid/webrtc/client/PeerTests.java | 37 --- .../webrtc/client/RuntimeCloseTests.java | 246 ++++++++++++++++++ .../webrtc/client/SerialExecutorTests.java | 199 ++++++++++++++ .../webrtc/client/TestHttpEndpoint.java | 106 +++++--- 8 files changed, 1091 insertions(+), 113 deletions(-) create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/ManualExecutor.java create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/ManualTimer.java create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java create mode 100644 webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java 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 f0e86970..3524311d 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -128,9 +128,11 @@ private WebRtc(Builder builder) { this.closeTimeout = builder.closeTimeout; this.errorHandler = builder.errorHandler != null ? builder.errorHandler : WebRtc::uncaught; - this.controlPool = Executors.newCachedThreadPool(threads("webrtc-client-control-" + runtime + "-")); + this.controlPool = builder.controlPool != null ? builder.controlPool + : Executors.newCachedThreadPool(threads("webrtc-client-control-" + runtime + "-")); this.framePool = Executors.newCachedThreadPool(threads("webrtc-client-frames-" + runtime + "-")); - this.timer = Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-timer-" + runtime + "-")); + this.timer = builder.timer != null ? builder.timer + : Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-timer-" + runtime + "-")); if (builder.eventExecutor != null) { this.eventExecutor = builder.eventExecutor; @@ -237,31 +239,7 @@ public Peer createPeer(PeerOptions options) { int number = peerCount.incrementAndGet(); String name = options.name != null ? options.name : "peer-" + number; - PeerImpl peer = new PeerImpl(this, options, name); - - // Registered before the runtime is checked again: a runtime that - // begins to close either finds the peer among those it closes and - // waits for it before freeing the factory, or is seen closing here. - peers.add(peer); - - if (closing.get()) { - peers.remove(peer); - throw new PeerClosedException("The runtime is closed"); - } - - try { - peer.open(); - } - catch (RuntimeException e) { - // Closed before the error goes out, so that nothing of the failed - // peer still runs when the runtime closes and frees WebRTC's threads. - // Closing removes the peer. Once the runtime has stopped its threads - // the close cannot run any more, and there is nothing left to wait for. - CompletableFuture.anyOf(peer.closeInternal(), closed).join(); - throw e; - } - - return peer; + return open(register(new PeerImpl(this, options, name))); } /** @@ -335,6 +313,41 @@ public void close() { } } + /** + * Registers a peer before the runtime is checked again: a runtime that + * begins to close either finds the peer among those it closes and waits + * for it before freeing the factory, or is seen closing here. + */ + PeerImpl register(PeerImpl peer) { + peers.add(peer); + + if (closing.get()) { + peers.remove(peer); + throw new PeerClosedException("The runtime is closed"); + } + + return peer; + } + + /** + * Creates the connection of a registered peer. + */ + PeerImpl open(PeerImpl peer) { + try { + peer.open(); + } + catch (RuntimeException e) { + // Closed before the error goes out, so that nothing of the failed + // peer still runs when the runtime closes and frees WebRTC's threads. + // Closing removes the peer. Once the runtime has stopped its threads + // the close cannot run any more, and there is nothing left to wait for. + CompletableFuture.anyOf(peer.closeInternal(), closed).join(); + throw e; + } + + return peer; + } + void reportError(Throwable error) { try { errorHandler.accept(error); @@ -545,6 +558,10 @@ public static final class Builder { private Consumer factoryCustomizer; + private ExecutorService controlPool; + + private ScheduledExecutorService timer; + private Builder() { } @@ -661,6 +678,25 @@ public Builder configureFactory(Consumer customiz return this; } + /** + * Replaces the pool the control tasks run on, so that tests decide + * when each of them runs. The runtime shuts it down when it closes. + */ + Builder controlPool(ExecutorService pool) { + this.controlPool = requireNonNull(pool, "pool"); + return this; + } + + /** + * Replaces the timer of timeouts and periodic tasks, so that tests + * decide when each of them fires. The runtime shuts it down when it + * closes. + */ + Builder timer(ScheduledExecutorService timer) { + this.timer = requireNonNull(timer, "timer"); + return this; + } + /** * @return the runtime. * diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java index 1c8c8fa5..924b1d62 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java @@ -28,10 +28,12 @@ import java.time.Duration; import java.util.Arrays; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.UnaryOperator; import dev.onvoid.webrtc.RTCSdpType; import dev.onvoid.webrtc.RTCSessionDescription; @@ -44,6 +46,9 @@ class HttpSessionTests extends ClientTestBase { + /** Longer than any test takes: the session times out only when a test fires it. */ + private static final Duration SESSION_TIMEOUT = Duration.ofSeconds(25); + private ScheduledExecutorService feeder; private CustomVideo video; @@ -136,17 +141,82 @@ void refusedOfferLeavesNothing() throws Exception { @Test void lateAnswerDeleted() throws Exception { - // The answer arrives after the session timed out. The server created - // a resource for it all the same, which must not be left there. + // The session times out while the offer is out. The server created a + // resource for it all the same, which must not be left there. + ManualTimer timer = new ManualTimer(); + WebRtc client = clientRuntime(builder -> builder.timer(timer)); + try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> "", 201)) { - endpoint.stallMillis = 700; + CountDownLatch answer = endpoint.holdAnswers(); + CompletableFuture session = Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(client); - ExecutionException failed = assertThrows(ExecutionException.class, - () -> Whep.from(endpoint.uri()).timeout(Duration.ofSeconds(1)).play(rtc) - .get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + endpoint.awaitRequest("POST /live/endpoint", TIMEOUT_SECONDS); + assertEquals(1, timer.fire(SESSION_TIMEOUT)); + // Failed before the answer arrives, which then meets a closed peer. + ExecutionException failed = assertThrows(ExecutionException.class, () -> await(session)); assertTrue(failed.getCause().getMessage().contains("within"), failed.getCause().getMessage()); - awaitRequest(endpoint, "DELETE /live/session/1"); + + answer.countDown(); + endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS); + + client.close(); + assertEquals(1, endpoint.count("DELETE /live/session/1")); + } + finally { + client.close(); + } + } + + @Test + void timeoutAfterAnswerDeletes() throws Exception { + // The session times out once the answer was read, before it is applied. + // Who owns the resource then is the timeout, not the request. + ManualExecutor control = new ManualExecutor(); + ManualTimer timer = new ManualTimer(); + WebRtc client = clientRuntime(builder -> builder.controlPool(control).timer(timer)); + + try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> answer(rtc.createPeer(PeerOptions.builder() + .negotiation(Negotiation.MANUAL) + .build()), offer), 201)) { + CompletableFuture> playing = CompletableFuture.supplyAsync( + () -> Whep.from(endpoint.uri()).timeout(SESSION_TIMEOUT).play(client)); + + // The client's side up to the answer: the peer is created, the + // offer is created and posted, and the answer read. The request + // runs on this thread, within the task that sends it. + control.runUntil(() -> endpoint.answered() > 0, TIMEOUT_SECONDS); + assertEquals(1, timer.fire(SESSION_TIMEOUT)); + control.runAll(); + + ExecutionException failed = assertThrows(ExecutionException.class, () -> await(await(playing))); + assertTrue(failed.getCause().getMessage().contains("within"), failed.getCause().getMessage()); + // Deleted on this thread, by the last tasks run. + assertEquals(1, endpoint.count("DELETE /live/session/1")); + } + finally { + closeStepping(client, control); + } + } + + @Test + void closeAfterRuntimeReported() throws Exception { + // A runtime that closed has no threads left to delete the session on. + WebRtc client = clientRuntime(builder -> builder); + + try (TestHttpEndpoint endpoint = sendingEndpoint()) { + Whep.Session session = await(Whep.from(endpoint.uri()).play(client)); + + client.close(); + await(session.closeAsync()); + + assertEquals(0, endpoint.count("DELETE /live/session/1")); + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertTrue(errors.get(0).getMessage().contains("left on the server"), errors.get(0).getMessage()); + errors.clear(); + } + finally { + client.close(); } } @@ -161,7 +231,7 @@ void oversizedAnswerRefused() throws Exception { assertTrue(failed.getCause().getMessage().contains("larger than"), failed.getCause().getMessage()); // The resource was named before the answer was refused. - awaitRequest(endpoint, "DELETE /live/session/1"); + endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS); } } @@ -208,6 +278,38 @@ void resourceKeepsTls() { () -> HttpSession.resourceOf(URI.create("http://media.example/whep"), "file:/etc/passwd")); } + /** + * An endpoint whose side of the session sends the video. + */ + private TestHttpEndpoint sendingEndpoint() throws Exception { + return new TestHttpEndpoint(offer -> { + Peer server = rtc.createPeer(PeerOptions.builder().negotiation(Negotiation.MANUAL).build()); + server.addTrack(video); + return answer(server, offer); + }, 201); + } + + /** + * A runtime for the client's side of a session, apart from the one the + * endpoint's side runs on, so that its threads can be held or stopped. + */ + private WebRtc clientRuntime(UnaryOperator options) { + return options.apply(WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add)) + .build(); + } + + /** + * Closes a runtime whose control tasks run only when the test runs them. + */ + static void closeStepping(WebRtc runtime, ManualExecutor control) throws Exception { + CompletableFuture closed = runtime.closeAsync(); + + control.runUntil(closed::isDone, TIMEOUT_SECONDS); + await(closed); + } + /** * An endpoint whose side of the session receives what is published. */ @@ -227,18 +329,6 @@ private void assertDeleteReported(String status) { errors.clear(); } - private static void awaitRequest(TestHttpEndpoint endpoint, String request) throws InterruptedException { - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS); - - while (endpoint.requests.stream().noneMatch(r -> r.toString().equals(request)) - && System.nanoTime() < deadline) { - Thread.sleep(50); - } - - assertTrue(endpoint.requests.stream().anyMatch(r -> r.toString().equals(request)), - () -> "Requests: " + endpoint.requests); - } - private static String answer(Peer peer, String offer) { try { return peer.negotiator() diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/ManualExecutor.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/ManualExecutor.java new file mode 100644 index 00000000..baecbcb2 --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/ManualExecutor.java @@ -0,0 +1,201 @@ +/* + * 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; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +/** + * An executor whose tasks run only when a test runs them, on the test's thread, + * so that the test decides how they interleave with everything else. Like a + * pool, it rejects tasks once it is shut down, and still runs those queued + * before; it can also be told to reject tasks while it runs. + */ +final class ManualExecutor extends AbstractExecutorService { + + private final ArrayDeque queue = new ArrayDeque<>(); + + private boolean shutdown; + + private boolean rejecting; + + private int running; + + + @Override + public synchronized void execute(Runnable task) { + if (shutdown) { + throw new RejectedExecutionException("Shut down"); + } + if (rejecting) { + throw new RejectedExecutionException("Rejecting"); + } + + queue.add(task); + notifyAll(); + } + + /** + * @param rejecting true to reject tasks as a full pool does. + */ + synchronized void rejecting(boolean rejecting) { + this.rejecting = rejecting; + } + + synchronized int queued() { + return queue.size(); + } + + /** + * Runs the oldest task queued. + * + * @return false if none was queued. + */ + boolean runNext() { + Runnable task; + + synchronized (this) { + task = queue.poll(); + + if (task == null) { + return false; + } + + running++; + } + + try { + task.run(); + } + finally { + synchronized (this) { + running--; + notifyAll(); + } + } + + return true; + } + + /** + * Runs tasks until none is queued, including those they queue. + * + * @return the number of tasks run. + */ + int runAll() { + int count = 0; + + while (runNext()) { + count++; + } + + return count; + } + + /** + * Waits until a task is queued, which tells that another thread got as far + * as handing it over. + */ + synchronized void awaitQueued(long timeoutSeconds) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + + while (queue.isEmpty()) { + long left = deadline - System.nanoTime(); + + if (left <= 0) { + throw new IllegalStateException("No task was queued within " + timeoutSeconds + " s"); + } + + TimeUnit.NANOSECONDS.timedWait(this, left); + } + } + + /** + * Runs tasks as they are queued until the condition holds. Other threads + * may queue them meanwhile. + */ + void runUntil(BooleanSupplier condition, long timeoutSeconds) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + + while (!condition.getAsBoolean()) { + if (runNext()) { + continue; + } + + synchronized (this) { + long left = deadline - System.nanoTime(); + + if (left <= 0) { + throw new IllegalStateException("The condition did not hold within " + timeoutSeconds + " s"); + } + if (queue.isEmpty()) { + // Woken by a new task; the condition may change without one. + TimeUnit.NANOSECONDS.timedWait(this, Math.min(left, TimeUnit.MILLISECONDS.toNanos(10))); + } + } + } + } + + @Override + public synchronized void shutdown() { + shutdown = true; + notifyAll(); + } + + @Override + public synchronized List shutdownNow() { + shutdown = true; + + List left = new ArrayList<>(queue); + queue.clear(); + notifyAll(); + + return left; + } + + @Override + public synchronized boolean isShutdown() { + return shutdown; + } + + @Override + public synchronized boolean isTerminated() { + return shutdown && queue.isEmpty() && running == 0; + } + + @Override + public synchronized boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + long deadline = System.nanoTime() + unit.toNanos(timeout); + + while (!isTerminated()) { + long left = deadline - System.nanoTime(); + + if (left <= 0) { + return false; + } + + TimeUnit.NANOSECONDS.timedWait(this, left); + } + + return true; + } +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/ManualTimer.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/ManualTimer.java new file mode 100644 index 00000000..c0d77192 --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/ManualTimer.java @@ -0,0 +1,195 @@ +/* + * 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; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Callable; +import java.util.concurrent.Delayed; +import java.util.concurrent.Executors; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +/** + * A timer on which no time passes: what is scheduled fires only when a test + * fires it, on the test's thread. Tasks are told apart by the delay they were + * scheduled with, so that a timeout fires exactly when the test wants it to, + * however long the work it bounds takes. + */ +final class ManualTimer extends AbstractExecutorService implements ScheduledExecutorService { + + private final List> tasks = new ArrayList<>(); + + private boolean shutdown; + + + /** + * Fires every task scheduled with the delay that has not been cancelled. + * Periodic tasks stay scheduled, the others are done. + * + * @return the number of tasks fired. + */ + int fire(Duration delay) { + List> due = new ArrayList<>(); + + synchronized (this) { + for (Task task : tasks) { + if (task.delayMillis == delay.toMillis() && !task.isDone()) { + due.add(task); + } + } + for (Task task : due) { + if (!task.periodic) { + tasks.remove(task); + } + } + } + + for (Task task : due) { + task.fire(); + } + + return due.size(); + } + + /** + * @return how many tasks with the delay wait to be fired. + */ + synchronized int scheduled(Duration delay) { + int count = 0; + + for (Task task : tasks) { + if (task.delayMillis == delay.toMillis() && !task.isDone()) { + count++; + } + } + + return count; + } + + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + return add(new Task<>(Executors.callable(command, null), unit.toMillis(delay), false)); + } + + @Override + public ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit) { + return add(new Task<>(callable, unit.toMillis(delay), false)); + } + + @Override + public ScheduledFuture scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { + return add(new Task<>(Executors.callable(command, null), unit.toMillis(period), true)); + } + + @Override + public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, + TimeUnit unit) { + return add(new Task<>(Executors.callable(command, null), unit.toMillis(delay), true)); + } + + @Override + public void execute(Runnable command) { + schedule(command, 0, TimeUnit.MILLISECONDS); + } + + @Override + public synchronized void shutdown() { + shutdown = true; + } + + @Override + public synchronized List shutdownNow() { + shutdown = true; + + List left = new ArrayList<>(); + + for (Task task : tasks) { + task.cancel(false); + left.add(task); + } + tasks.clear(); + + return left; + } + + @Override + public synchronized boolean isShutdown() { + return shutdown; + } + + @Override + public synchronized boolean isTerminated() { + return shutdown; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return isTerminated(); + } + + private synchronized Task add(Task task) { + if (shutdown) { + throw new RejectedExecutionException("Shut down"); + } + + tasks.add(task); + + return task; + } + + + + private static final class Task extends FutureTask implements ScheduledFuture { + + final long delayMillis; + + final boolean periodic; + + + Task(Callable callable, long delayMillis, boolean periodic) { + super(callable); + + this.delayMillis = delayMillis; + this.periodic = periodic; + } + + void fire() { + if (periodic) { + runAndReset(); + } + else { + run(); + } + } + + @Override + public long getDelay(TimeUnit unit) { + return unit.convert(delayMillis, TimeUnit.MILLISECONDS); + } + + @Override + public int compareTo(Delayed other) { + return Long.compare(getDelay(TimeUnit.MILLISECONDS), other.getDelay(TimeUnit.MILLISECONDS)); + } + } +} 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 7e86e880..4002429c 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -440,43 +440,6 @@ 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(); - } - } - @Test void sendBeyondLimitFails() throws Exception { // Never negotiated, so the channel stays connecting and holds every send. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java new file mode 100644 index 00000000..940dd68a --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java @@ -0,0 +1,246 @@ +/* + * 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; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +/** + * Closes runtimes at each point where it can meet another operation. The + * runtimes run their control tasks, or their events, on a {@link + * ManualExecutor}, so that each test decides the order rather than the + * scheduler. + */ +class RuntimeCloseTests extends ClientTestBase { + + @Test + void closeBeforeRegister() throws Exception { + // createPeer checked the runtime, then the runtime began to close and + // found no peer to wait for. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = runtime(control); + PeerImpl peer = new PeerImpl(runtime, manualOptions(), "late"); + + CompletableFuture closed = runtime.closeAsync(); + + assertThrows(PeerClosedException.class, () -> runtime.register(peer)); + assertFalse(runtime.peers().contains(peer)); + + // Nothing of the peer reached the control executor, so no connection + // was created, and the runtime closes without it. + assertEquals(0, control.queued()); + await(closed); + } + + @Test + void closeBeforeOpen() throws Exception { + // The peer is registered, and the runtime closes before the connection + // is created. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = runtime(control); + PeerImpl peer = runtime.register(new PeerImpl(runtime, manualOptions(), "registered")); + + CompletableFuture closed = runtime.closeAsync(); + + // The runtime found the peer and queued its teardown, and waits for it + // before it frees the factory. + control.awaitQueued(TIMEOUT_SECONDS); + assertFalse(closed.isDone()); + + // Whether the creation joins the teardown's turn or takes one of its + // own, the teardown runs first. + CompletableFuture opened = CompletableFuture.supplyAsync(() -> runtime.open(peer)); + control.runUntil(opened::isDone, TIMEOUT_SECONDS); + + ExecutionException e = assertThrows(ExecutionException.class, () -> await(opened)); + assertInstanceOf(PeerClosedException.class, e.getCause()); + + control.runUntil(closed::isDone, TIMEOUT_SECONDS); + await(closed); + assertEquals(PeerState.CLOSED, peer.state()); + assertTrue(runtime.peers().isEmpty()); + } + + @Test + void closeWhileOpening() throws Exception { + // The task that creates the connection is queued when the runtime + // begins to close. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = runtime(control); + + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.awaitQueued(TIMEOUT_SECONDS); + + CompletableFuture closed = runtime.closeAsync(); + assertFalse(closed.isDone()); + + control.runUntil(closed::isDone, TIMEOUT_SECONDS); + await(closed); + + // Created first, so createPeer returns it, then closed with the runtime. + Peer peer = await(created); + assertEquals(PeerState.CLOSED, peer.state()); + await(peer.whenClosed()); + } + + @Test + void closeTimeoutReported() throws Exception { + // The peer's teardown never runs, so neither close finishes. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .controlPool(control) + .closeTimeout(Duration.ofMillis(100)) + .build(); + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + await(created); + + try { + runtime.close(); + + // The runtime cannot tell whether WebRTC still uses the factory, + // so it leaves it. That outlasts close(), whose wait ran out first. + ExecutionException e = assertThrows(ExecutionException.class, + () -> runtime.closeAsync().get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + assertEquals(2, errors.size(), () -> "Errors: " + errors); + assertTrue(errors.stream().anyMatch(error -> error.getMessage().contains("did not close within")), + () -> "Errors: " + errors); + assertTrue(errors.contains(e.getCause()), () -> "Errors: " + errors); + assertTrue(e.getCause().getMessage().contains("left undisposed"), e.getCause()::getMessage); + } + finally { + // Lets the teardown close the connection. + control.runAll(); + errors.clear(); + } + } + + @Test + void closeOnEventThread() throws Exception { + // close() does not wait for the events of the peers it closes, which + // would wait for the very thread it runs on. + Peer peer = rtc.createPeer(manualOptions()); + CompletableFuture closedState = new CompletableFuture<>(); + + peer.onStateChange(state -> { + if (state == PeerState.NEW) { + rtc.close(); + closedState.complete(peer.state()); + } + }); + + assertEquals(PeerState.CLOSED, await(closedState)); + await(peer.whenClosed()); + // No "did not close within" either; the base class checks. + } + + @Test + void rejectedCloseEvents() throws Exception { + // The event executor refuses the events that tell of the close. The + // peer's futures settle all the same, on the thread that found them + // refused. + ManualExecutor events = new ManualExecutor(); + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .eventExecutor(events) + .build(); + + try { + Peer peer = runtime.createPeer(manualOptions()); + CompletableFuture connected = peer.whenConnected(); + + events.runAll(); + events.rejecting(true); + peer.close(); + + await(peer.whenClosed()); + ExecutionException e = assertThrows(ExecutionException.class, () -> await(connected)); + assertInstanceOf(PeerClosedException.class, e.getCause()); + + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertInstanceOf(RejectedExecutionException.class, errors.get(0).getCause()); + } + finally { + events.rejecting(false); + runtime.close(); + events.runAll(); + errors.clear(); + } + } + + @Test + void closeFailureReportedOnce() throws Exception { + // What closing fails with reaches the error handler once, whether the + // application waits through close() or closeAsync(). + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .controlPool(control) + .closeTimeout(Duration.ofMillis(100)) + .build(); + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + await(created); + + try { + ExecutionException e = assertThrows(ExecutionException.class, + () -> runtime.closeAsync().get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + runtime.close(); + runtime.close(); + + assertInstanceOf(WebRtcException.class, e.getCause()); + assertTrue(e.getCause().getMessage().contains("left undisposed"), e.getCause()::getMessage); + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertSame(e.getCause(), errors.get(0)); + } + finally { + control.runAll(); + errors.clear(); + } + } + + private WebRtc runtime(ManualExecutor control) { + return WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .controlPool(control) + .build(); + } + + private static PeerOptions manualOptions() { + return PeerOptions.builder().negotiation(Negotiation.MANUAL).build(); + } + +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java new file mode 100644 index 00000000..298e67f0 --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/SerialExecutorTests.java @@ -0,0 +1,199 @@ +/* + * 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; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.RejectedExecutionException; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import org.junit.jupiter.api.Test; + +/** + * Drives a {@link SerialExecutor} through a {@link ManualExecutor}, so that + * each test decides when the pool runs a task and when it rejects one. + */ +class SerialExecutorTests { + + private final List errors = new ArrayList<>(); + + private final ManualExecutor pool = new ManualExecutor(); + + private final SerialExecutor serial = new SerialExecutor(pool, errors::add, false, null); + + + @Test + void rejectionRunsFallback() { + List ran = new ArrayList<>(); + SerialExecutor refused = new SerialExecutor(task -> { + throw new RejectedExecutionException("Full"); + }, errors::add, false, null); + + refused.execute(() -> ran.add("task"), () -> ran.add("fallback on " + Thread.currentThread().getName())); + + // On the thread that found the task rejected, before execute returns. + assertEquals(Collections.singletonList("fallback on " + Thread.currentThread().getName()), ran); + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertInstanceOf(WebRtcException.class, errors.get(0)); + assertInstanceOf(RejectedExecutionException.class, errors.get(0).getCause()); + } + + @Test + void shutdownRejectsQuietly() { + List ran = new ArrayList<>(); + pool.shutdown(); + + serial.execute(() -> ran.add("task"), () -> ran.add("fallback")); + serial.execute(() -> ran.add("plain")); + + // The pool's owner shut it down and fails what it loses itself. + assertEquals(Arrays.asList("fallback"), ran); + assertTrue(errors.isEmpty(), () -> "Errors: " + errors); + } + + @Test + void rejectedTurnDropsRest() { + List ran = new ArrayList<>(); + List dropped = new ArrayList<>(); + int count = 70; + + for (int i = 0; i < count; i++) { + int n = i; + + if (n == 66) { + // One without a fallback is dropped as well, and counted. + serial.execute(() -> ran.add(n)); + continue; + } + serial.execute(() -> ran.add(n), () -> dropped.add(n)); + } + + // One turn on the pool for all of them. + assertEquals(1, pool.queued()); + + // The pool fills up while the first batch runs, so the next turn is refused. + pool.rejecting(true); + pool.runNext(); + + assertEquals(range(0, 64), ran); + assertEquals(Arrays.asList(64, 65, 67, 68, 69), dropped); + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertTrue(errors.get(0).getMessage().contains("rejected 6 tasks"), errors.get(0).getMessage()); + + // Taken again once the pool accepts work. + pool.rejecting(false); + serial.execute(() -> ran.add(100)); + pool.runAll(); + + assertEquals(100, (int) ran.get(ran.size() - 1)); + } + + @Test + void throwingFallbackReported() { + List dropped = new ArrayList<>(); + IllegalStateException failure = new IllegalStateException("fallback failed"); + + for (int i = 0; i < 64; i++) { + serial.execute(() -> { }); + } + serial.execute(() -> { }, () -> { + throw failure; + }); + serial.execute(() -> { }, () -> dropped.add(1)); + + pool.rejecting(true); + pool.runNext(); + + // The fallback after the one that threw still runs. + assertEquals(Collections.singletonList(1), dropped); + assertEquals(2, errors.size(), () -> "Errors: " + errors); + assertSame(failure, errors.get(1)); + } + + @Test + void batchesShareThePool() { + List ran = new ArrayList<>(); + SerialExecutor other = new SerialExecutor(pool, errors::add, false, null); + + for (int i = 0; i < 100; i++) { + int n = i; + serial.execute(() -> ran.add("serial-" + n)); + } + other.execute(() -> ran.add("other")); + + pool.runNext(); + pool.runNext(); + pool.runNext(); + + // The other executor's task runs after the first batch, not after all. + assertEquals("serial-63", ran.get(63)); + assertEquals("other", ran.get(64)); + assertEquals("serial-64", ran.get(65)); + assertEquals(101, ran.size()); + assertEquals(0, pool.queued()); + } + + @Test + void throwingTaskReported() { + List ran = new ArrayList<>(); + IllegalStateException failure = new IllegalStateException("task failed"); + + serial.execute(() -> { + throw failure; + }); + serial.execute(() -> ran.add("next")); + pool.runAll(); + + assertEquals(Collections.singletonList("next"), ran); + assertEquals(Collections.singletonList(failure), errors); + } + + @Test + void eventThreadMarked() { + SerialExecutor events = new SerialExecutor(pool, errors::add, true, "events-test"); + List seen = new ArrayList<>(); + String name = Thread.currentThread().getName(); + + events.execute(() -> { + seen.add(SerialExecutor.onEventThread()); + seen.add(events.inExecutor()); + seen.add(serial.inExecutor()); + seen.add(Thread.currentThread().getName()); + }); + pool.runAll(); + + assertEquals(Arrays.asList(true, true, false, "events-test"), seen); + assertFalse(SerialExecutor.onEventThread()); + assertFalse(events.inExecutor()); + assertEquals(name, Thread.currentThread().getName()); + assertTrue(errors.isEmpty(), () -> "Errors: " + errors); + } + + private static List range(int from, int to) { + return IntStream.range(from, to).boxed().collect(Collectors.toList()); + } + +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java index 439b3f28..033682e4 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/TestHttpEndpoint.java @@ -28,12 +28,15 @@ import java.util.List; import java.util.Locale; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.function.Function; /** * Just enough HTTP for WHIP and WHEP tests. A POST is answered with what the * handler returns for the offer, and creates a resource; a DELETE of that - * resource returns 200, unless the test sets another status. + * resource returns 200, unless the test sets another status. Requests are + * served one at a time, in the order they arrive. */ final class TestHttpEndpoint implements AutoCloseable { @@ -70,12 +73,10 @@ public String toString() { /** Where a DELETE answered with a redirect points. */ volatile String redirect = "/live/moved"; - /** - * How long the answer to a POST pauses, twice: before its status line and - * after it. Each pause stays within a client's read timeout while both - * together may not. - */ - volatile long stallMillis; + /** Holds the answers to POSTs while set, until it is released. */ + private volatile CountDownLatch answerGate; + + private int answered; private final ServerSocket server; @@ -102,6 +103,49 @@ URI uri() { return URI.create("http://127.0.0.1:" + server.getLocalPort() + "/live/endpoint"); } + /** + * Holds the answer to every POST from now on, after the request has been + * recorded, until the returned latch is counted down. + */ + CountDownLatch holdAnswers() { + CountDownLatch gate = new CountDownLatch(1); + answerGate = gate; + return gate; + } + + /** + * @return how many POSTs have been answered, counted before the answer is + * written: once a client has read one, it is counted. + */ + synchronized int answered() { + return answered; + } + + /** + * Waits until the endpoint has received the request, such as "DELETE + * /live/session/1". + */ + synchronized void awaitRequest(String request, long timeoutSeconds) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + + while (requests.stream().noneMatch(r -> r.toString().equals(request))) { + long left = deadline - System.nanoTime(); + + if (left <= 0) { + throw new AssertionError("No " + request + " within " + timeoutSeconds + " s. Requests: " + requests); + } + + TimeUnit.NANOSECONDS.timedWait(this, left); + } + } + + /** + * @return how many of the requests received are the given one. + */ + long count(String request) { + return requests.stream().filter(r -> r.toString().equals(request)).count(); + } + @Override public void close() throws IOException { server.close(); @@ -143,52 +187,56 @@ else if (name.equals("authorization")) { read += in.read(body, read, length - read); } - requests.add(new Request(requestLine[0], requestLine[1], authorization)); + synchronized (this) { + requests.add(new Request(requestLine[0], requestLine[1], authorization)); + notifyAll(); + } OutputStream out = socket.getOutputStream(); if (requestLine[0].equals("POST")) { if (status != 201) { - respond(out, status + " Refused", "", "not allowed", 0); + respond(out, status + " Refused", "", "not allowed"); return; } String answer = answerer.apply(new String(body, StandardCharsets.UTF_8)); - respond(out, "201 Created", "Location: " + location + "\r\n", answer, stallMillis); + CountDownLatch gate = answerGate; + + if (gate != null) { + try { + gate.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + + synchronized (this) { + answered++; + } + + respond(out, "201 Created", "Location: " + location + "\r\n", answer); } else if (deleteStatus / 100 == 3) { - respond(out, deleteStatus + " Moved", "Location: " + redirect + "\r\n", "", 0); + respond(out, deleteStatus + " Moved", "Location: " + redirect + "\r\n", ""); } else { - respond(out, deleteStatus + " Status", "", "", 0); + respond(out, deleteStatus + " Status", "", ""); } } - private static void respond(OutputStream out, String status, String headers, String body, long stallMillis) - throws IOException { + private static void respond(OutputStream out, String status, String headers, String body) throws IOException { byte[] bytes = body.getBytes(StandardCharsets.UTF_8); - String head = "Content-Type: application/sdp\r\n" + headers + String head = "HTTP/1.1 " + status + "\r\nContent-Type: application/sdp\r\n" + headers + "Content-Length: " + bytes.length + "\r\nConnection: close\r\n\r\n"; - stall(stallMillis); - out.write(("HTTP/1.1 " + status + "\r\n").getBytes(StandardCharsets.US_ASCII)); - out.flush(); - stall(stallMillis); out.write(head.getBytes(StandardCharsets.US_ASCII)); out.write(bytes); out.flush(); } - private static void stall(long millis) throws IOException { - try { - Thread.sleep(millis); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IOException(e); - } - } - private static String readLine(InputStream in) throws IOException { ByteArrayOutputStream line = new ByteArrayOutputStream(); int c;