diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java index 44eeaec20..7bf302f03 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java @@ -375,47 +375,59 @@ void slowFrameHandlerGetsNewest() throws Exception { Peer[] peers = pair(); CustomVideo video = rtc.media().customVideo("video"); AtomicInteger handled = new AtomicInteger(); - CompletableFuture enough = new CompletableFuture<>(); + CompletableFuture busy = new CompletableFuture<>(); + CompletableFuture resumed = new CompletableFuture<>(); + CountDownLatch done = new CountDownLatch(1); peers[1].onTrack(track -> track.asVideo().onFrame(frame -> { - try { - // Much slower than the 30 frames per second sent. - Thread.sleep(200); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); + int count = handled.incrementAndGet(); + + if (count == 3) { + // Busy until the test lets it go, however fast frames come. + busy.complete(null); + + try { + done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } - if (handled.incrementAndGet() == 3) { - enough.complete(null); + else if (count > 3) { + resumed.complete(null); } })); peers[0].addTrack(video); feedVideo(video, 160, 120); - await(enough); + await(busy); - // The decoder keeps going while the handler is busy: within the same - // time it gets well ahead of the handler, at whatever rate the encoder - // has ramped up to on this machine. + // The decoder keeps going while the handler is busy, at whatever rate + // the encoder has ramped up to on this machine, and the handler gets + // none of those frames meanwhile. long decodedBefore = framesDecoded(await(peers[1].stats())); - int handledBefore = handled.get(); long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(TIMEOUT_SECONDS); long decodedMeanwhile; - int handledMeanwhile; - do { - Thread.sleep(100); - decodedMeanwhile = framesDecoded(await(peers[1].stats())) - decodedBefore; - handledMeanwhile = handled.get() - handledBefore; - } - while ((decodedMeanwhile < 10 || decodedMeanwhile < handledMeanwhile + 5) && System.nanoTime() < deadline); + try { + do { + Thread.sleep(100); + decodedMeanwhile = framesDecoded(await(peers[1].stats())) - decodedBefore; + } + while (decodedMeanwhile < 10 && System.nanoTime() < deadline); + + long decoded = decodedMeanwhile; - long decoded = decodedMeanwhile; - int handledFrames = handledMeanwhile; + assertTrue(decoded >= 10, () -> "Meanwhile decoded " + decoded + " frames"); + assertEquals(3, handled.get()); + } + finally { + done.countDown(); + } - assertTrue(decoded >= 10 && decoded >= handledFrames + 5, - () -> "Meanwhile decoded " + decoded + " frames and handled " + handledFrames); + // Then gets the newest frame, and the ones after it. + await(resumed); } private static long framesDecoded(PeerStats stats) { diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java index 9d6217499..b845370f0 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java @@ -25,6 +25,7 @@ import java.time.Duration; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; @@ -153,9 +154,24 @@ void closeWaitsForTasksQueuedBehindTeardown() throws Exception { void closeTimeoutReported() throws Exception { // The peer's teardown never runs, so neither close finishes. ManualExecutor control = new ManualExecutor(); + CountDownLatch closeReturned = new CountDownLatch(1); WebRtc runtime = WebRtc.builder() .audio(AudioMode.HEADLESS) - .errorHandler(errors::add) + .errorHandler(error -> { + errors.add(error); + + // Closing ends only after close() has returned. Otherwise a + // close() that wakes up late finds it ended and has no + // timeout to report. + if (error.getMessage().contains("left undisposed")) { + try { + closeReturned.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }) .controlPool(control) .closeTimeout(Duration.ofMillis(100)) .build(); @@ -164,7 +180,12 @@ void closeTimeoutReported() throws Exception { await(created); try { - runtime.close(); + try { + runtime.close(); + } + finally { + closeReturned.countDown(); + } // The runtime cannot tell whether WebRTC still uses the factory, // so it leaves it. That outlasts close(), whose wait ran out first. diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/internal/SinkRegistryTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/internal/SinkRegistryTests.java index 20fd98a1e..fb84703ac 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/internal/SinkRegistryTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/internal/SinkRegistryTests.java @@ -169,19 +169,21 @@ void removeInCallbackIsDeferred() throws Exception { return add().add(s, r); }); - assertTrue(registration.get().enter()); + try (HeldChanges held = holdDeferredChanges()) { + assertTrue(registration.get().enter()); - try { - // WebRTC holds its lock here, so the native removal must wait. - registry.remove(sink); + try { + // WebRTC holds its lock here, so the native removal must wait. + registry.remove(sink); - assertEquals(List.of("add 1"), calls); + assertEquals(List.of("add 1"), calls); - // Removed for the Java side at once. - assertFalse(registration.get().enter()); - } - finally { - registration.get().exit(); + // Removed for the Java side at once. + assertFalse(registration.get().enter()); + } + finally { + registration.get().exit(); + } } SinkRegistry.awaitDeferred(); @@ -194,16 +196,18 @@ void addInCallbackIsDeferred() throws Exception { SinkRegistry.Registration callback = callback(); Object sink = new Object(); - assertTrue(callback.enter()); + try (HeldChanges held = holdDeferredChanges()) { + assertTrue(callback.enter()); - try { - registry.add(sink, add()); - registry.add(sink, add()); + try { + registry.add(sink, add()); + registry.add(sink, add()); - assertEquals(List.of("add 1"), calls); - } - finally { - callback.exit(); + assertEquals(List.of("add 1"), calls); + } + finally { + callback.exit(); + } } SinkRegistry.awaitDeferred(); @@ -216,14 +220,16 @@ void addAndRemoveInCallback() throws Exception { SinkRegistry.Registration callback = callback(); Object sink = new Object(); - assertTrue(callback.enter()); + try (HeldChanges held = holdDeferredChanges()) { + assertTrue(callback.enter()); - try { - registry.add(sink, add()); - registry.remove(sink); - } - finally { - callback.exit(); + try { + registry.add(sink, add()); + registry.remove(sink); + } + finally { + callback.exit(); + } } SinkRegistry.awaitDeferred(); @@ -294,6 +300,47 @@ private SinkRegistry.Registration callback() { return registration.get(); } + /** + * Holds the thread that makes the native calls deferred from callbacks, + * so that those a test defers wait until it closes the returned hold: + * otherwise that thread may make them before the test looks. + */ + private static HeldChanges holdDeferredChanges() throws InterruptedException { + HeldChanges held = new HeldChanges(); + AtomicReference callback = new AtomicReference<>(); + + new SinkRegistry<>(handle -> { }).add(new Object(), (s, r) -> { + callback.set(r); + return 1; + }); + + assertTrue(callback.get().enter()); + + try { + // Deferred, and made first: the deferred calls are made one after + // another, so the ones deferred later wait for this one. + new SinkRegistry<>(handle -> { }).add(new Object(), (s, r) -> { + held.entered.countDown(); + + try { + assertTrue(held.released.await(5, TimeUnit.SECONDS)); + } + catch (InterruptedException e) { + throw new IllegalStateException(e); + } + + return 1; + }); + } + finally { + callback.get().exit(); + } + + assertTrue(held.entered.await(5, TimeUnit.SECONDS)); + + return held; + } + private SinkRegistry.NativeAdd add() { return (sink, registration) -> { long handle = nextHandle.incrementAndGet(); @@ -304,6 +351,21 @@ private SinkRegistry.NativeAdd add() { + /** The thread of the deferred native calls, held until closed. */ + private static final class HeldChanges implements AutoCloseable { + + private final CountDownLatch entered = new CountDownLatch(1); + + private final CountDownLatch released = new CountDownLatch(1); + + + @Override + public void close() { + released.countDown(); + } + + } + /** A native add that waits, once it has been entered, until the test lets it return. */ private final class BlockedAdd implements SinkRegistry.NativeAdd {