From c3a6dcb808beb04d0a3831894a40395f666b78d5 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Fri, 23 Oct 2026 17:41:32 +0200 Subject: [PATCH] fix: keep camera ownership consistent and fail switches peers did not take A switch that lets go of the camera first takes the release from the track, and ends the track when neither camera opens again; the camera cache only drops an entry that is still the camera released. A sender that cannot take the new track, or does not in time, now fails the switch, which goes back to the old camera while it is still open. Replaced tracks are released once no sender uses them. --- docs/guide/client-api.md | 2 +- .../dev/onvoid/webrtc/client/LocalCamera.java | 4 + .../onvoid/webrtc/client/LocalCameraImpl.java | 234 ++++++++++++++++-- .../onvoid/webrtc/client/LocalMediaImpl.java | 3 +- .../onvoid/webrtc/client/LocalTrackImpl.java | 15 +- .../dev/onvoid/webrtc/client/SenderImpl.java | 15 +- .../webrtc/client/CameraIntegrationTest.java | 117 +++++++++ 7 files changed, 357 insertions(+), 33 deletions(-) diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 16c2b760..0ca4fac4 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -402,7 +402,7 @@ camera.switchCamera(media.devices().cameras().get(1)) .thenRun(() -> System.out.println("Now sending " + camera.device().getName())); ``` -The new camera keeps the resolution and frame rate of the old one, as far as it supports them. The switch opens the new camera before it closes the old one, so that the peers see no gap. Some devices cannot capture with two cameras at once, such as the two sensors of a laptop camera; then the old camera closes first, and the switch takes about half a second. If the new camera cannot be opened, the future fails with a `MediaException`, and the track keeps sending the camera it had. +The new camera keeps the resolution and frame rate of the old one, as far as it supports them. The switch opens the new camera before it closes the old one, so that the peers see no gap. Some devices cannot capture with two cameras at once, such as the two sensors of a laptop camera; then the old camera closes first, and the switch takes about half a second. If the new camera cannot be opened, the future fails with a `MediaException`, and the track keeps sending the camera it had. If a peer does not take the new camera within the operation timeout, the future fails with a `WebRtcException`; the track goes back to the old camera, or keeps the new one when the old one had to close first. A track that closed its camera first and can open neither camera ends. ### Devices diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCamera.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCamera.java index 421097c3..24175f1b 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCamera.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCamera.java @@ -41,6 +41,10 @@ public interface LocalCamera extends LocalVideo { * @return a future that completes once every peer sends the new camera. It * fails with a {@link MediaException} if the camera cannot be opened, or is * already open with other options, and the track keeps its camera then. + * It fails with a {@link WebRtcException} if a peer does not take the new + * camera within the operation timeout; the track goes back to its camera + * then, or keeps the new one if the device had to close the old one first. + * A track that closed its camera first and can open neither one ends. */ CompletableFuture switchCamera(VideoDevice device); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java index 23f22d4f..4665021e 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalCameraImpl.java @@ -22,7 +22,9 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import dev.onvoid.webrtc.media.video.VideoDevice; import dev.onvoid.webrtc.media.video.VideoTrack; @@ -34,9 +36,13 @@ */ final class LocalCameraImpl extends LocalVideoImpl implements LocalCamera { - /** Guarded by this. */ + // Guarded by this. + private LocalMediaImpl.SharedCamera camera; + /** Replaced native tracks with their cameras, which late senders may still send. */ + private final List retiring = new ArrayList<>(); + LocalCameraImpl(LocalMediaImpl media, VideoTrack track, LocalMediaImpl.SharedCamera camera) { super(media, track, () -> media.releaseCamera(camera)); @@ -59,7 +65,8 @@ public CompletableFuture switchCamera(VideoDevice device) { // Opening a camera can take a moment, so it happens off the caller's thread. media.executor.execute(() -> { try { - switchTo(device, switched); + switchTo(device); + switched.complete(null); } catch (Throwable e) { switched.completeExceptionally(e instanceof WebRtcException @@ -70,7 +77,7 @@ public CompletableFuture switchCamera(VideoDevice device) { return switched; } - private void switchTo(VideoDevice device, CompletableFuture switched) { + private void switchTo(VideoDevice device) { LocalMediaImpl.SharedCamera current; synchronized (this) { @@ -78,14 +85,11 @@ private void switchTo(VideoDevice device, CompletableFuture switched) { } if (current.options.device.equals(device)) { - switched.complete(null); return; } CameraOptions target = current.options.withDevice(device); LocalMediaImpl.SharedCamera next; - boolean currentReleased = false; - MediaException failure = null; try { // Opened first, so that peers see no gap. @@ -94,13 +98,36 @@ private void switchTo(VideoDevice device, CompletableFuture switched) { catch (MediaException beside) { // Some cameras cannot capture at the same time, such as two // sensors of one device. Unless another track shares the current - // camera, it is let go first, and comes back if the new one fails. + // camera, it is let go first. if (!media.isOnlyUser(current)) { throw beside; } - media.releaseCamera(current); - currentReleased = true; + switchInstead(current, target); + return; + } + + WebRtcException notTaken = replace(next, true); + + if (notTaken != null) { + throw notTaken; + } + } + + /** + * Switches after letting go of the current camera, which comes back if + * the new one fails. A track left without either camera ends. + */ + private void switchInstead(LocalMediaImpl.SharedCamera current, CameraOptions target) { + // The track no longer owns the camera, so that it does not let go of + // it a second time when it is released. + takeRelease().run(); + + MediaException failure = null; + WebRtcException notTaken; + + try { + LocalMediaImpl.SharedCamera next; try { next = media.openCamera(target); @@ -109,24 +136,45 @@ private void switchTo(VideoDevice device, CompletableFuture switched) { failure = instead; next = media.openCamera(current.options); } + + notTaken = replace(next, false); } + catch (RuntimeException e) { + closeInternal(); - replace(next, currentReleased); + MediaException ended = new MediaException("The track let go of the camera " + + current.options.device.getName() + " to switch, got no camera back, and ended", e); + + if (failure != null && failure != e) { + ended.addSuppressed(failure); + } + + throw ended; + } if (failure != null) { - switched.completeExceptionally(failure); + if (notTaken != null) { + failure.addSuppressed(notTaken); + } + + throw failure; } - else { - switched.complete(null); + if (notTaken != null) { + throw notTaken; } } /** * Puts a track over the camera in place of the current one, on this - * track and on every sender, and lets the old one go once no sender sends - * it any more. + * track and on every sender, and lets the old one go once every sender + * sends the new one. When not every sender does in time, the track goes + * back to the old camera if it can, and keeps the new one otherwise. + * + * @param canGoBack whether the old camera is still open. + * + * @return null once every sender sends the new track, else why not. */ - private void replace(LocalMediaImpl.SharedCamera next, boolean oldCameraReleased) { + private WebRtcException replace(LocalMediaImpl.SharedCamera next, boolean canGoBack) { VideoTrack nextTrack; try { @@ -138,9 +186,11 @@ private void replace(LocalMediaImpl.SharedCamera next, boolean oldCameraReleased + " could not be created", e); } + LocalMediaImpl.SharedCamera previous; LocalTrackImpl.Replaced replaced; synchronized (this) { + previous = camera; replaced = replaceNative(nextTrack, () -> media.releaseCamera(next)); if (replaced != null) { @@ -156,23 +206,159 @@ private void replace(LocalMediaImpl.SharedCamera next, boolean oldCameraReleased throw new MediaException("The camera track is closed"); } + CompletableFuture sent = send(replaced.senders); + WebRtcException notTaken = await(sent, "Not every peer took the camera " + next.options.device.getName()); + + if (notTaken != null && canGoBack && goBack(replaced, previous, sent)) { + return notTaken; + } + + retire(replaced, sent); + + return notTaken; + } + + /** + * Puts the replaced track and camera back, on this track and on every + * sender, and lets go of the new ones. + * + * @return false if the track closed meanwhile, whose release then lets go + * of the new ones. + */ + private boolean goBack(LocalTrackImpl.Replaced replaced, LocalMediaImpl.SharedCamera previous, + CompletableFuture sent) { + LocalTrackImpl.Replaced undone; + + synchronized (this) { + undone = replaceNative(replaced.track, replaced.release); + + if (undone != null) { + moveSinks((VideoTrack) undone.track); + camera = previous; + } + } + + if (undone == null) { + return false; + } + + // A sender that is late sends the old track anyway, as it reads the + // track when its task runs. + CompletableFuture back = send(undone.senders); + WebRtcException notBack = await(back, + "Not every peer went back to the camera " + previous.options.device.getName()); + + if (notBack != null) { + rtc.reportError(notBack); + } + + retire(undone, CompletableFuture.allOf(sent, back)); + + return true; + } + + /** + * Lets go of a replaced native track and its camera once no sender uses + * them any more, or when this track is released, if that comes first. + * Late senders are done by then, as the senders detach from this track + * on the same executors. + */ + private void retire(LocalTrackImpl.Replaced replaced, CompletableFuture sent) { + if (sent.isDone()) { + retireNow(replaced); + return; + } + + synchronized (this) { + retiring.add(replaced); + } + + sent.whenComplete((ignored, error) -> media.executor.execute(() -> { + boolean pending; + + synchronized (this) { + pending = retiring.remove(replaced); + } + + if (pending) { + retireNow(replaced); + } + })); + } + + private void retireNow(LocalTrackImpl.Replaced replaced) { + try { + letGo(replaced); + } + catch (Throwable e) { + rtc.reportError(new WebRtcException("Releasing the camera track replaced failed", e)); + } + } + + @Override + void beforeRelease() { + List left; + + synchronized (this) { + left = new ArrayList<>(retiring); + retiring.clear(); + } + + Throwable failure = attempt(null, super::beforeRelease); + + for (LocalTrackImpl.Replaced replaced : left) { + failure = attempt(failure, () -> letGo(replaced)); + } + + if (failure instanceof RuntimeException) { + throw (RuntimeException) failure; + } + if (failure instanceof Error) { + throw (Error) failure; + } + } + + /** + * Makes the senders send the native track this track has now. + */ + private CompletableFuture send(List senders) { List> sending = new ArrayList<>(); - for (SenderImpl sender : replaced.senders) { + for (SenderImpl sender : senders) { sending.add(sender.trackReplaced(this)); } + return CompletableFuture.allOf(sending.toArray(new CompletableFuture[0])); + } + + /** + * @return null once the senders send within the operation timeout, else + * why not. + */ + private WebRtcException await(CompletableFuture sent, String failure) { try { - CompletableFuture.allOf(sending.toArray(new CompletableFuture[0])) - .get(rtc.operationTimeout.toMillis(), TimeUnit.MILLISECONDS); + sent.get(rtc.operationTimeout.toMillis(), TimeUnit.MILLISECONDS); + + return null; } - catch (Exception e) { - rtc.reportError(new WebRtcException("Not every peer took the new camera in time", e)); + catch (ExecutionException e) { + return new WebRtcException(failure, e.getCause()); } + catch (TimeoutException e) { + return new WebRtcException(failure + " within " + rtc.operationTimeout, e); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return new WebRtcException(failure, e); + } + } - replaced.track.dispose(); - - if (!oldCameraReleased) { + private static void letGo(LocalTrackImpl.Replaced replaced) { + try { + replaced.track.dispose(); + } + finally { + // The camera holds a reference of its own, so it goes even when the track does not. replaced.release.run(); } } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java index c5bc0c23..58ad24e0 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java @@ -379,7 +379,8 @@ void releaseCamera(SharedCamera camera) { return; } - cameras.remove(camera.key); + // Another camera of the same device may have been opened since. + cameras.remove(camera.key, camera); } camera.source.stop(); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java index 552d529e..60c74cb9 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalTrackImpl.java @@ -164,6 +164,19 @@ synchronized Replaced replaceNative(MediaStreamTrack next, Runnable nextRelease) return replaced; } + /** + * Takes the release of the source from the track, which then no longer + * lets go of the source, even once it ended. + * + * @return the release taken. + */ + synchronized Runnable takeRelease() { + Runnable taken = releaseSource; + releaseSource = () -> { }; + + return taken; + } + synchronized boolean register(SenderImpl sender) { if (ended) { return false; @@ -272,7 +285,7 @@ private void release() { * * @return the failure so far, with the step's added to it. */ - private static Throwable attempt(Throwable failure, Runnable step) { + static Throwable attempt(Throwable failure, Runnable step) { try { step.run(); } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java index b634389c..f52469d5 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java @@ -287,24 +287,27 @@ CompletableFuture trackClosed(LocalTrackImpl closedTrack) { * Sends the native track a local track uses now, after it replaced the one * before, without renegotiating. * - * @return a future that completes once the sender sends the new track. + * @return a future that completes once the sender sends the new track, or + * no longer sends the local track, and fails if it could not take it. */ CompletableFuture trackReplaced(LocalTrackImpl replacedIn) { CompletableFuture done = new CompletableFuture<>(); peer.control.execute(() -> { try { + // Reads the native track now, so that a late task sends the + // one the local track has then. if (!removed && sender != null && track == replacedIn && !peer.isClosed()) { sender.replaceTrack(replacedIn.nativeTrack()); } + + done.complete(null); } catch (Throwable e) { - peer.rtc.reportError(e); + done.completeExceptionally(PeerImpl.asWebRtcException(e, + "The sender of " + peer.name + " could not take the new track")); } - finally { - done.complete(null); - } - }); + }, () -> done.completeExceptionally(new PeerClosedException("The runtime is closed"))); return done; } diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/CameraIntegrationTest.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/CameraIntegrationTest.java index 069b5729..a786fc26 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/CameraIntegrationTest.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/CameraIntegrationTest.java @@ -17,14 +17,20 @@ package dev.onvoid.webrtc.client; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assumptions.assumeTrue; +import java.time.Duration; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -130,6 +136,117 @@ void switchesCamera() throws Exception { waitFor(() -> atOne.get() > oneAfter + 10, "frames after the failed switch"); } + @Test + void latePeerFailsSwitch() throws Exception { + List cameras = cameras(2); + + rtc.close(); + rtc = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .operationTimeout(Duration.ofSeconds(2)) + .build(); + + Peer[] peers = pair(); + LocalCamera camera = rtc.media().camera(CameraOptions.builder().device(cameras.get(0)).build()); + AtomicInteger received = new AtomicInteger(); + + // Some cameras cannot capture at the same time. Then the switch lets + // go of the old one first, and cannot go back to it. + boolean together; + + try { + await(rtc.media().camera(CameraOptions.builder().device(cameras.get(1)).build()).closeAsync()); + together = true; + } + catch (MediaException e) { + together = false; + } + + peers[1].onTrack(track -> track.asVideo().onFrame(frame -> received.incrementAndGet())); + peers[0].addTrack(camera); + waitFor(() -> received.get() > 5, "frames before the switch"); + + // The peer's control executor is busy, so the peer takes neither the + // new camera, nor the old one back, in time. + CountDownLatch busy = new CountDownLatch(1); + ((PeerImpl) peers[0]).control.execute(() -> { + try { + busy.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + ExecutionException failed = assertThrows(ExecutionException.class, + () -> camera.switchCamera(cameras.get(1)).get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertInstanceOf(WebRtcException.class, failed.getCause()); + assertInstanceOf(TimeoutException.class, failed.getCause().getCause()); + assertFalse(camera.isEnded()); + + if (together) { + assertEquals(cameras.get(0), camera.device()); + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertTrue(errors.remove(0).getMessage().startsWith("Not every peer went back")); + } + else { + assertEquals(cameras.get(1), camera.device()); + } + + // Once free, the peer sends the camera the track has. The track it + // sent before is released then, which the runtime's close checks. + busy.countDown(); + int before = received.get(); + waitFor(() -> received.get() > before + 10, "frames after the failed switch"); + } + + @Test + void lostCameraEndsTrack() throws Exception { + VideoDevice device = cameras(1).get(0); + CameraOptions first = CameraOptions.builder().device(device).resolution(640, 480).build(); + CameraOptions second = CameraOptions.builder().device(device).resolution(320, 240).build(); + AtomicReference other = new AtomicReference<>(); + + LocalCamera camera = rtc.media().camera(first); + + // The switch lets go of the camera as the missing one cannot be opened + // beside it. Meanwhile another track takes the camera with other + // options, so the switch cannot open it again. + VideoDevice missing = new VideoDevice("Missing Camera", "/dev/video-missing") { + + @Override + public String getDescriptor() { + if (other.get() == null) { + try { + other.set(rtc.media().camera(second)); + } + catch (MediaException stillOpen) { + // Not let go of yet. + } + } + + return super.getDescriptor(); + } + }; + + ExecutionException failed = assertThrows(ExecutionException.class, + () -> camera.switchCamera(missing).get(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertInstanceOf(MediaException.class, failed.getCause()); + assertTrue(camera.isEnded()); + assertNotNull(other.get()); + + // Closing the ended track lets go of nothing the other track uses. + await(camera.closeAsync()); + + CompletableFuture frame = new CompletableFuture<>(); + other.get().onFrame(video -> frame.complete(video.buffer.getWidth())); + assertEquals(320, (int) await(frame)); + + // The capture of the other track is still the one shared. + assertThrows(MediaException.class, () -> rtc.media().camera(first)); + } + private List cameras(int needed) { List cameras = rtc.media().devices().cameras(); assumeTrue(cameras.size() >= needed, "Needs " + needed + " camera(s), found " + cameras.size());