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 cde25cb0..cf47f0bd 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whep.java @@ -170,8 +170,15 @@ public CompletableFuture play(WebRtc rtc) { PeerImpl peer = (PeerImpl) rtc.createPeer(options.build()); - return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, bearerToken, timeout) - .thenApply(resource -> new Session(rtc, peer, resource, bearerToken, timeout))); + try { + return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, bearerToken, timeout) + .thenApply(resource -> new Session(rtc, peer, resource, bearerToken, timeout))); + } + catch (RuntimeException | Error e) { + // Until the session has it, no one else would close the peer. + peer.close(); + throw e; + } } } } 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 044b2b0e..5414858a 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/Whip.java @@ -147,7 +147,9 @@ public Builder timeout(Duration timeout) { * fails if the endpoint refuses the offer or the peer does not connect * in time, with nothing left behind on the server. * - * @throws IllegalStateException if no track was given. + * @throws IllegalStateException if no track was given. + * @throws IllegalArgumentException if a track was not created by the + * runtime. */ public CompletableFuture publish(WebRtc rtc) { requireNonNull(rtc, "rtc"); @@ -156,6 +158,13 @@ public CompletableFuture publish(WebRtc rtc) { throw new IllegalStateException("Give at least one track to publish"); } + // Before the peer exists, so that a wrong track leaves nothing behind. + for (LocalTrack track : tracks) { + if (!(track instanceof LocalTrackImpl) || ((LocalTrackImpl) track).rtc != rtc) { + throw new IllegalArgumentException("The track " + track + " was not created by the runtime"); + } + } + PeerOptions.Builder options = PeerOptions.builder() .name("whip-" + endpoint.getHost()) .negotiation(Negotiation.MANUAL); @@ -166,12 +175,19 @@ public CompletableFuture publish(WebRtc rtc) { PeerImpl peer = (PeerImpl) rtc.createPeer(options.build()); - for (LocalTrack track : tracks) { - peer.addTrack(track); - } + try { + for (LocalTrack track : tracks) { + peer.addTrack(track); + } - return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, bearerToken, timeout) - .thenApply(resource -> new Session(rtc, peer, resource, bearerToken, timeout))); + return peer.toPublic(HttpSession.negotiate(rtc, peer, endpoint, bearerToken, timeout) + .thenApply(resource -> new Session(rtc, peer, resource, bearerToken, timeout))); + } + catch (RuntimeException | Error e) { + // Until the session has it, no one else would close the peer. + peer.close(); + throw e; + } } } } 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 924b1d62..a9f3167c 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/HttpSessionTests.java @@ -31,6 +31,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.UnaryOperator; @@ -139,6 +140,48 @@ void refusedOfferLeavesNothing() throws Exception { } } + @Test + void foreignTrackLeavesNothing() throws Exception { + WebRtc other = clientRuntime(UnaryOperator.identity()); + + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + Whip.Builder whip = Whip.to(endpoint.uri()).track(video).track(other.media().customVideo("other")); + + assertThrows(IllegalArgumentException.class, () -> whip.publish(rtc)); + assertTrue(rtc.peers().isEmpty(), () -> "Peers left: " + rtc.peers()); + assertTrue(endpoint.requests.isEmpty()); + } + finally { + other.close(); + } + } + + @Test + void failedStartClosesPeer() throws Exception { + // The session's deadline cannot be scheduled, after the peer was created. + ManualTimer timer = new ManualTimer(); + WebRtc client = clientRuntime(builder -> builder.timer(timer)); + + try (TestHttpEndpoint endpoint = receivingEndpoint()) { + CustomVideo track = client.media().customVideo("video"); + timer.shutdown(); + + assertThrows(RejectedExecutionException.class, + () -> Whip.to(endpoint.uri()).track(track).publish(client)); + assertThrows(RejectedExecutionException.class, () -> Whep.from(endpoint.uri()).play(client)); + + for (Peer peer : client.peers()) { + PeerImpl impl = (PeerImpl) peer; + assertTrue(impl.isClosing(), () -> "Not closing: " + peer); + impl.closeInternal().get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + assertTrue(client.peers().isEmpty(), () -> "Peers left: " + client.peers()); + } + finally { + client.close(); + } + } + @Test void lateAnswerDeleted() throws Exception { // The session times out while the offer is out. The server created a