diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index f25c0b9f..e1926e33 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -543,7 +543,7 @@ Whep.Session viewer = Whep.from(URI.create("https://example.com/live/whep")) viewer.peer().onTrack(track -> System.out.println("Receiving " + track.kind())); ``` -The offer goes to the endpoint with the local candidates, as WHIP and WHEP servers expect, and the future completes once the peer is connected. If the endpoint refuses the offer, or the peer does not connect within the timeout (15 seconds by default), the future fails, and the session is deleted on the server again. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none. +The offer goes to the endpoint with the local candidates, as WHIP and WHEP servers expect, and the future completes once the peer is connected. If the endpoint refuses the offer, or the peer does not connect within the timeout (15 seconds by default), the future fails, and the session is deleted on the server again. Cancelling the future, say once `get` gave up waiting, ends the session the same way, even when the answer is still on its way. `onTrack` hands over tracks that already arrived, so a handler added once the future completed misses none. `session.close()` deletes the session on the server, which frees what the server holds for it, and closes the peer. The tracks a WHIP session published stay open. `session.peer()` is a regular peer, for stats and anything else. A deletion the server refuses or that fails goes to the runtime's error handler. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java index 974ff1ed..f001003f 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/HttpSession.java @@ -26,6 +26,7 @@ import java.net.URI; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledFuture; @@ -122,14 +123,41 @@ public void close() { closeAsync(); } + /** + * Negotiates the peer over HTTP and hands the session out once the peer is + * connected. A caller that stops waiting, by cancelling the future or + * completing it itself, ends the negotiation, and a session made too late + * for the future is closed: either way the peer is closed and what the + * server created is deleted. + * + * @param factory makes the session of the connected peer. + * + * @return the future handed to the application. + */ + static CompletableFuture start(WebRtc rtc, PeerImpl peer, URI endpoint, + String bearerToken, Duration timeout, Factory factory) { + CompletableFuture abandoned = new CompletableFuture<>(); + CompletableFuture session = peer.toPublic(negotiate(rtc, peer, endpoint, bearerToken, timeout, abandoned) + .thenApply(resource -> factory.create(rtc, peer, resource, bearerToken, timeout)), + HttpSession::close); + + // Once the session was handed out, this finds the negotiation finished. + session.whenComplete((value, error) -> abandoned.complete(null)); + + return session; + } + /** * Negotiates a manual peer over HTTP: creates the offer with all candidates, * posts it, applies the answer and waits until the peer is connected. * + * @param abandoned completes when no one waits for the session anymore, + * which then ends as if it had failed. + * * @return a future with the resource the server named, or null. */ - static CompletableFuture negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken, - Duration timeout) { + private static CompletableFuture negotiate(WebRtc rtc, Peer peer, URI endpoint, String bearerToken, + Duration timeout, CompletableFuture abandoned) { CompletableFuture result = new CompletableFuture<>(); // The resource the server created. Whoever takes it out owns it: the // session that connected, or whoever deletes it. @@ -164,6 +192,11 @@ static CompletableFuture negotiate(WebRtc rtc, Peer peer, URI endpoint, Str "The session with " + endpoint + " was not connected within " + timeout)), timeout.toMillis(), TimeUnit.MILLISECONDS); + abandoned.thenRun(() -> { + deadline.cancel(false); + fail.accept(new CancellationException("The session with " + endpoint + " was abandoned")); + }); + peer.negotiator().createOffer(Gathering.COMPLETE) .thenApplyAsync(offer -> { try { @@ -238,6 +271,16 @@ private static String post(URI endpoint, String sdp, String bearerToken, Duratio } } + /** + * Makes the session of a peer that connected. + * + * @param the session type. + */ + interface Factory { + + S create(WebRtc rtc, Peer peer, URI resource, String bearerToken, Duration timeout); + } + private static void delete(URI resource, String bearerToken, Duration timeout) throws IOException { HttpURLConnection connection = open(resource, "DELETE", bearerToken, timeout); 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 62735b49..e840959b 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -564,14 +564,32 @@ long channelReceiveLimit() { * executor, so its callbacks run there like the handlers. */ CompletableFuture toPublic(CompletableFuture internal) { + return toPublic(internal, value -> { }); + } + + /** + * Hands a future out to the application, as {@link #toPublic(CompletableFuture)} + * does, and frees a result that comes too late for it: the caller may have + * cancelled or completed the future meanwhile, and then no one else would. + * + * @param unclaimed frees a result the future did not take. + */ + CompletableFuture toPublic(CompletableFuture internal, Consumer unclaimed) { ClientFuture result = new ClientFuture<>(); rtc.track(result, events); - // Completed in any case: should the event executor reject the task, on - // the thread that found it rejected. - internal.whenComplete((value, error) -> events.execute(() -> complete(result, value, error), - () -> complete(result, value, error))); + internal.whenComplete((value, error) -> { + Runnable deliver = () -> { + if (!complete(result, value, error) && error == null && value != null) { + unclaimed.accept(value); + } + }; + + // Completed in any case: should the event executor reject the + // task, on the thread that found it rejected. + events.execute(deliver, deliver); + }); return result; } @@ -681,13 +699,15 @@ static WebRtcException asWebRtcException(Throwable error, String message) { return new WebRtcException(message + ": " + error, error); } - static void complete(CompletableFuture future, T value, Throwable error) { + /** + * @return whether this completed the future, which was not done before. + */ + static boolean complete(CompletableFuture future, T value, Throwable error) { if (error != null) { - future.completeExceptionally(unwrap(error)); - } - else { - future.complete(value); + return future.completeExceptionally(unwrap(error)); } + + return future.complete(value); } private LocalTrackImpl checkTrack(LocalTrack track) { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java index 4c84fc50..244b2965 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java @@ -154,7 +154,8 @@ public Builder timeout(Duration timeout) { * * @return a future with the session once its peer is connected. It * fails if the endpoint refuses the offer or the peer does not connect - * in time, with nothing left behind on the server. + * in time, with nothing left behind on the server. Cancelling it ends + * the session the same way. */ public CompletableFuture play(WebRtc rtc) { requireNonNull(rtc, "rtc"); @@ -175,8 +176,7 @@ public CompletableFuture play(WebRtc rtc) { PeerImpl peer = (PeerImpl) rtc.createPeer(options.build()); try { - return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, token, limit) - .thenApply(resource -> new Session(rtc, peer, resource, token, limit))); + return HttpSession.start(rtc, peer, endpoint, token, limit, Session::new); } catch (RuntimeException | Error e) { // Until the session has it, no one else would close the peer. diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java index 981b276f..4561e91e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java @@ -145,7 +145,8 @@ public Builder timeout(Duration timeout) { * * @return a future with the session once its peer is connected. It * fails if the endpoint refuses the offer or the peer does not connect - * in time, with nothing left behind on the server. + * in time, with nothing left behind on the server. Cancelling it ends + * the session the same way. * * @throws IllegalStateException if no track was given. * @throws IllegalArgumentException if a track was not created by the @@ -184,8 +185,7 @@ public CompletableFuture publish(WebRtc rtc) { peer.addTrack(track); } - return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, token, limit) - .thenApply(resource -> new Session(rtc, peer, resource, token, limit))); + return HttpSession.start(rtc, peer, endpoint, token, limit, Session::new); } catch (RuntimeException | Error e) { // Until the session has it, no one else would close the peer. 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 6d9babf8..956e61aa 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java @@ -34,6 +34,7 @@ import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import java.util.function.UnaryOperator; import dev.onvoid.webrtc.RTCSdpType; @@ -211,6 +212,54 @@ void lateAnswerDeleted() throws Exception { } } + @Test + void cancelledStartDeletes() throws Exception { + // The caller cancels while the offer is out. The peer must not go on to + // connect, and what the server creates must not be left there. + WebRtc client = clientRuntime(builder -> builder); + + try { + CustomVideo track = client.media().customVideo("video"); + + assertCancelDeletes(client, uri -> Whip.to(uri).track(track).publish(client)); + assertCancelDeletes(client, uri -> Whep.from(uri).play(client)); + } + finally { + client.close(); + } + } + + @Test + void unclaimedSessionClosed() throws Exception { + // The caller cancels once the peer is connected, as the session is + // made or right before. A session the future cannot take anymore is + // closed, rather than left connected with no one to close it. + WebRtc client = clientRuntime(builder -> builder); + + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + CountDownLatch answer = endpoint.holdAnswers(); + CompletableFuture session = Whip.to(endpoint.uri()) + .track(client.media().customVideo("video")) + .publish(client); + + endpoint.awaitRequest("POST /live/endpoint", TIMEOUT_SECONDS); + Peer peer = client.peers().get(0); + CompletableFuture cancelled = peer.whenConnected().thenRun(() -> session.cancel(true)); + + answer.countDown(); + await(cancelled); + await(peer.whenClosed()); + endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS); + + assertTrue(session.isCancelled()); + 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. @@ -377,6 +426,32 @@ private WebRtc clientRuntime(UnaryOperator options) { .build(); } + /** + * Starts a session, cancels it while the answer is held and releases the + * answer, whose resource must then be deleted. + */ + private static void assertCancelDeletes(WebRtc client, + Function> start) throws Exception { + try (TestHttpEndpoint endpoint = new TestHttpEndpoint(offer -> "", 201)) { + CountDownLatch answer = endpoint.holdAnswers(); + CompletableFuture session = start.apply(endpoint.uri()); + + endpoint.awaitRequest("POST /live/endpoint", TIMEOUT_SECONDS); + PeerImpl peer = (PeerImpl) client.peers().get(0); + + assertTrue(session.cancel(true)); + // Closed by the cancellation itself, so it never connects. + assertTrue(peer.isClosing(), () -> "Not closing: " + peer); + + answer.countDown(); + endpoint.awaitRequest("DELETE /live/session/1", TIMEOUT_SECONDS); + await(peer.whenClosed()); + + assertTrue(session.isCancelled()); + assertEquals(1, endpoint.count("DELETE /live/session/1")); + } + } + /** * Closes a runtime whose control tasks run only when the test runs them. */ 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 9686b49d..685f5cf9 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/PeerTests.java @@ -64,6 +64,28 @@ void channelRoundTrip() throws Exception { assertTrue(errors.isEmpty(), () -> "Errors: " + errors); } + @Test + void lateResultFreed() throws Exception { + // A result that comes after the caller cancelled has no one else to + // free it. One the caller takes is left alone. + PeerImpl peer = (PeerImpl) rtc.createPeer(PeerOptions.builder().negotiation(Negotiation.MANUAL).build()); + CompletableFuture late = new CompletableFuture<>(); + CompletableFuture freed = new CompletableFuture<>(); + CompletableFuture cancelled = peer.toPublic(late, freed::complete); + + assertTrue(cancelled.cancel(true)); + late.complete("late"); + assertEquals("late", await(freed)); + + CompletableFuture taken = new CompletableFuture<>(); + CompletableFuture kept = new CompletableFuture<>(); + CompletableFuture handed = peer.toPublic(taken, kept::complete); + + taken.complete("taken"); + assertEquals("taken", await(handed)); + assertFalse(kept.isDone()); + } + @Test void stateHandlerGetsCurrentState() throws Exception { Peer[] peers = pair();