diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java index e840959b..c829d873 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -30,6 +30,7 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -752,13 +753,13 @@ private CompletableFuture enqueue(Supplier> operatio CompletableFuture current = running; runningOperation = current; - ScheduledFuture timeout = rtc.timer.schedule(() -> control.execute(() -> - current.completeExceptionally(new TimeoutException("WebRTC did not finish an operation of " - + name + " within " + rtc.operationTimeout))), - rtc.operationTimeout.toMillis(), TimeUnit.MILLISECONDS); + // One that is done already, as on a closed peer, needs no timeout. + ScheduledFuture timeout = current.isDone() ? null : timeOut(current); current.whenComplete((value, error) -> control.execute(() -> { - timeout.cancel(false); + if (timeout != null) { + timeout.cancel(false); + } runningOperation = null; operationRunning = false; @@ -773,6 +774,25 @@ private CompletableFuture enqueue(Supplier> operatio return result; } + /** + * Fails a running operation once the operation timeout has passed. + * + * @return the timeout, or null once the runtime has stopped its timer. + */ + private ScheduledFuture timeOut(CompletableFuture operation) { + try { + return rtc.timer.schedule(() -> control.execute(() -> + operation.completeExceptionally(new TimeoutException("WebRTC did not finish an operation of " + + name + " within " + rtc.operationTimeout))), + rtc.operationTimeout.toMillis(), TimeUnit.MILLISECONDS); + } + catch (RejectedExecutionException e) { + // The runtime is closing: control tasks queued before still run, + // and it fails the futures they leave open. + return null; + } + } + private void runNextOperation() { if (operationRunning) { return; @@ -907,11 +927,18 @@ private CompletableFuture waitForGathering() { // A STUN or TURN server that cannot be reached holds up the end of // gathering for long; the description then goes with what there is. - rtc.timer.schedule(() -> control.execute(() -> { - if (gatheringWaiters.remove(future)) { - future.complete(null); - } - }), options.gatheringTimeout.toMillis(), TimeUnit.MILLISECONDS); + try { + rtc.timer.schedule(() -> control.execute(() -> { + if (gatheringWaiters.remove(future)) { + future.complete(null); + } + }), options.gatheringTimeout.toMillis(), TimeUnit.MILLISECONDS); + } + catch (RejectedExecutionException e) { + // The runtime is closing and has no timer left to wait with. + gatheringWaiters.remove(future); + future.complete(null); + } return future; } @@ -1029,15 +1056,20 @@ private void watchCollision() { int answers = remoteAnswers; - collisionCheck = rtc.timer.schedule(() -> control.execute(() -> { - collisionCheck = null; + try { + collisionCheck = rtc.timer.schedule(() -> control.execute(() -> { + collisionCheck = null; - if (!closed && remoteAnswers == answers - && connection.getSignalingState() == RTCSignalingState.HAVE_LOCAL_OFFER) { - deliverError(new NegotiationException("Both peers sent an offer at once, and both are impolite, " - + "so neither gives way. Build exactly one of them with PeerOptions.Builder.polite(true)")); - } - }), COLLISION_TIMEOUT_MS, TimeUnit.MILLISECONDS); + if (!closed && remoteAnswers == answers + && connection.getSignalingState() == RTCSignalingState.HAVE_LOCAL_OFFER) { + deliverError(new NegotiationException("Both peers sent an offer at once, and both are impolite, " + + "so neither gives way. Build exactly one of them with PeerOptions.Builder.polite(true)")); + } + }), COLLISION_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } + catch (RejectedExecutionException e) { + // The runtime is closing, and the peer with it. + } } // What WebRTC reports, on the control executor. 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 b5efc66d..55d5c747 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java @@ -30,6 +30,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import dev.onvoid.webrtc.RTCSdpType; +import dev.onvoid.webrtc.RTCSessionDescription; import dev.onvoid.webrtc.media.audio.AudioDevice; import dev.onvoid.webrtc.media.video.CustomVideoSource; @@ -360,6 +362,68 @@ void switchDeviceAfterClose() throws Exception { await(closed); } + @Test + void closedPeerNeedsNoTimer() throws Exception { + // An operation reaches a peer the closing runtime has closed, after it + // stopped its timer, as control tasks queued before still run. + ManualExecutor control = new ManualExecutor(); + ManualTimer timer = new ManualTimer(); + WebRtc runtime = runtime(control, timer); + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + Peer peer = await(created); + + try { + CompletableFuture peerClosed = peer.closeAsync(); + control.runUntil(peerClosed::isDone, TIMEOUT_SECONDS); + + CompletableFuture operation = peer.withNative(connection -> null); + timer.shutdown(); + control.runUntil(operation::isDone, TIMEOUT_SECONDS); + + ExecutionException e = assertThrows(ExecutionException.class, () -> await(operation)); + assertInstanceOf(PeerClosedException.class, e.getCause()); + assertTrue(errors.isEmpty(), () -> "Errors: " + errors); + } + finally { + HttpSessionTests.closeStepping(runtime, control); + } + } + + @Test + void operationWithoutTimer() throws Exception { + // A peer the runtime could not close still runs what was queued, with + // no timer left to bound how long WebRTC or gathering takes. + ManualExecutor control = new ManualExecutor(); + ManualTimer timer = new ManualTimer(); + WebRtc runtime = runtime(control, timer); + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + Peer peer = await(created); + + try { + timer.shutdown(); + + CompletableFuture offer = peer.negotiator().createOffer(Gathering.COMPLETE); + control.runUntil(offer::isDone, TIMEOUT_SECONDS); + + assertEquals(RTCSdpType.OFFER, await(offer).sdpType); + assertTrue(errors.isEmpty(), () -> "Errors: " + errors); + } + finally { + HttpSessionTests.closeStepping(runtime, control); + } + } + + private WebRtc runtime(ManualExecutor control, ManualTimer timer) { + return WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .controlPool(control) + .timer(timer) + .build(); + } + private WebRtc runtime(ManualExecutor control) { return WebRtc.builder() .audio(AudioMode.HEADLESS)