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;