test: make three timing-dependent tests deterministic

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<Void> enough = new CompletableFuture<>();
+		CompletableFuture<Void> busy = new CompletableFuture<>();
+		CompletableFuture<Void> 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) {

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.

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<SinkRegistry.Registration> 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<Object> add() {
 		return (sink, registration) -> {
 			long handle = nextHandle.incrementAndGet();
@@ -304,6 +351,21 @@ private SinkRegistry.NativeAdd<Object> 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<Object> {