diff --git a/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/recorder/MediaRecorder.java b/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/recorder/MediaRecorder.java index eee94cb5..f3df8cc7 100644 --- a/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/recorder/MediaRecorder.java +++ b/webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/recorder/MediaRecorder.java @@ -316,7 +316,7 @@ public void close() { /** Called with the lock held, in the idle state. */ private void addTrack(EncodedFrames.Source source) { - addTrack(source.frames(), source::requestKeyFrame); + addTrack(source.take(), source::requestKeyFrame); } /** @@ -324,11 +324,12 @@ private void addTrack(EncodedFrames.Source source) { * control thread never takes this recorder's lock, so waiting under it * cannot deadlock. */ - private static EncodedFrames.Source sourceOf(CompletableFuture future) { + static EncodedFrames.Source sourceOf(CompletableFuture future) { try { return future.get(10, TimeUnit.SECONDS); } catch (InterruptedException e) { + giveUp(future); Thread.currentThread().interrupt(); throw new IllegalStateException("Interrupted while adding the track", e); } @@ -336,10 +337,21 @@ private static EncodedFrames.Source sourceOf(CompletableFuture future) { + future.cancel(false); + future.thenAccept(EncodedFrames.Source::close); + } + private void addTrack(long frames, Runnable keyFrameRequest) { if (frames == 0) { throw new IllegalArgumentException("The sender or receiver was disposed"); diff --git a/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/recorder/ClientApiRecorderTest.java b/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/recorder/ClientApiRecorderTest.java index 6061fafd..d663c6f3 100644 --- a/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/recorder/ClientApiRecorderTest.java +++ b/webrtc-java-media/src/test/java/dev/onvoid/webrtc/media/recorder/ClientApiRecorderTest.java @@ -43,6 +43,7 @@ import dev.onvoid.webrtc.client.Signaling; import dev.onvoid.webrtc.client.SignalingPair; import dev.onvoid.webrtc.client.WebRtc; +import dev.onvoid.webrtc.client.internal.EncodedFrames; import dev.onvoid.webrtc.media.player.MediaInfo; import dev.onvoid.webrtc.media.player.MediaReader; import dev.onvoid.webrtc.media.video.NativeI420Buffer; @@ -210,6 +211,31 @@ void removedSenderRefused() throws Exception { } } + @Test + void interruptedWaitCancels() { + // The peer has not handed the source over yet; once it does, it finds + // the future cancelled and closes the source itself. + CompletableFuture pending = new CompletableFuture<>(); + + Thread.currentThread().interrupt(); + + assertThrows(IllegalStateException.class, () -> MediaRecorder.sourceOf(pending)); + assertTrue(Thread.interrupted()); + assertTrue(pending.isCancelled()); + } + + @Test + void sourceBeforeGivingUpClosed() throws Exception { + // The source arrived just as the recorder stopped waiting, so no + // recorder takes it over, and giving up drops its reference. + CompletableFuture frames = EncodedFrames.of(videoSender); + EncodedFrames.Source source = frames.get(10, TimeUnit.SECONDS); + + MediaRecorder.giveUp(frames); + + assertThrows(IllegalStateException.class, source::take); + } + private interface Tracks { void addTo(MediaRecorder recorder); diff --git a/webrtc-jni/src/main/cpp/include/JNI_NativeApi.h b/webrtc-jni/src/main/cpp/include/JNI_NativeApi.h index 3014d693..664d7713 100644 --- a/webrtc-jni/src/main/cpp/include/JNI_NativeApi.h +++ b/webrtc-jni/src/main/cpp/include/JNI_NativeApi.h @@ -54,6 +54,14 @@ extern "C" { JNIEXPORT jlong JNICALL Java_dev_onvoid_webrtc_internal_NativeApi_receiverFrames (JNIEnv *, jclass, jlong); + /* + * Class: dev_onvoid_webrtc_internal_NativeApi + * Method: releaseFrames + * Signature: (J)V + */ + JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_internal_NativeApi_releaseFrames + (JNIEnv *, jclass, jlong); + #ifdef __cplusplus } #endif diff --git a/webrtc-jni/src/main/cpp/src/JNI_NativeApi.cpp b/webrtc-jni/src/main/cpp/src/JNI_NativeApi.cpp index fced0ee9..0cc208a8 100644 --- a/webrtc-jni/src/main/cpp/src/JNI_NativeApi.cpp +++ b/webrtc-jni/src/main/cpp/src/JNI_NativeApi.cpp @@ -56,3 +56,9 @@ JNIEXPORT jlong JNICALL Java_dev_onvoid_webrtc_internal_NativeApi_receiverFrames return receiver != nullptr ? Retained(jni::EncodedFrameTransformer::Of(receiver)) : 0; } + +JNIEXPORT void JNICALL Java_dev_onvoid_webrtc_internal_NativeApi_releaseFrames +(JNIEnv * env, jclass caller, jlong frames) +{ + jni::GetExtensionApi()->encoded_frames_release(reinterpret_cast(frames)); +} 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 14785f1e..62735b49 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java @@ -576,6 +576,44 @@ CompletableFuture toPublic(CompletableFuture internal) { return result; } + /** + * Hands a native extension a source of encoded frames, taken on the + * control executor. The future fails once the runtime has stopped the + * threads that would take it, and a source that comes too late for the + * future, which the caller cancelled meanwhile, is closed rather than + * left with its reference. + * + * @param take takes the source, or throws why it cannot. + */ + CompletableFuture encodedFrames(Supplier take) { + CompletableFuture result = new CompletableFuture<>(); + + rtc.track(result, null); + + control.execute(() -> { + if (result.isDone()) { + // The caller stopped waiting before this ran. + return; + } + + try { + EncodedFrames.Source source = take.get(); + + if (!result.complete(source)) { + source.close(); + } + } + catch (Throwable e) { + result.completeExceptionally(e); + } + }, () -> { + // A runtime that stopped its threads dropped the task. + result.completeExceptionally(closedError()); + }); + + return result; + } + void deliverError(WebRtcException error) { events.execute(() -> { if (errorHandlers.isEmpty()) { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java index 09669e34..c8f160f7 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java @@ -157,32 +157,23 @@ void onControl(Runnable change) { * too. */ CompletableFuture encodedFrames() { - CompletableFuture result = new CompletableFuture<>(); + return peer.encodedFrames(() -> { + if (ended) { + throw new IllegalStateException("The track " + id + " ended"); + } - peer.control.execute(() -> { - try { - if (ended) { - throw new IllegalStateException("The track " + id + " ended"); - } + long frames = NativeApi.encodedFramesOf(receiver); - long frames = NativeApi.encodedFramesOf(receiver); + if (frames == 0) { + throw new IllegalStateException("The track " + id + " ended"); + } - if (frames == 0) { - throw new IllegalStateException("The track " + id + " ended"); + return new EncodedFrames.Source(frames, () -> onControl(() -> { + if (kind == MediaKind.VIDEO) { + receiver.requestKeyFrame(); } - - result.complete(new EncodedFrames.Source(frames, () -> onControl(() -> { - if (kind == MediaKind.VIDEO) { - receiver.requestKeyFrame(); - } - }))); - } - catch (Throwable e) { - result.completeExceptionally(e); - } + })); }); - - return result; } /** Detaches everything from the native track, on the control executor. */ 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..843981b9 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java @@ -314,30 +314,21 @@ CompletableFuture trackReplaced(LocalTrackImpl replacedIn) { * peer's control executor, where key frames are asked for too. */ CompletableFuture encodedFrames() { - CompletableFuture result = new CompletableFuture<>(); + return peer.encodedFrames(() -> { + checkUsable(); - peer.control.execute(() -> { - try { - checkUsable(); + long frames = NativeApi.encodedFramesOf(sender); - long frames = NativeApi.encodedFramesOf(sender); + if (frames == 0) { + throw new IllegalStateException("The sender was removed"); + } - if (frames == 0) { - throw new IllegalStateException("The sender was removed"); + return new EncodedFrames.Source(frames, () -> peer.control.execute(() -> { + if (kind == MediaKind.VIDEO && !removed && sender != null && !peer.isClosed()) { + sender.generateKeyFrame(); } - - result.complete(new EncodedFrames.Source(frames, () -> peer.control.execute(() -> { - if (kind == MediaKind.VIDEO && !removed && sender != null && !peer.isClosed()) { - sender.generateKeyFrame(); - } - }))); - } - catch (Throwable e) { - result.completeExceptionally(e); - } + })); }); - - return result; } /** diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/internal/EncodedFrames.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/internal/EncodedFrames.java index e7c8266c..d191a3ff 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/internal/EncodedFrames.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/internal/EncodedFrames.java @@ -19,9 +19,11 @@ import static java.util.Objects.requireNonNull; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicLong; import dev.onvoid.webrtc.client.RemoteTrack; import dev.onvoid.webrtc.client.Sender; +import dev.onvoid.webrtc.internal.NativeApi; /** * Lets a native extension, such as the media module's recorder, observe the @@ -32,6 +34,11 @@ * them on the control executor of their peer. What this hands out keeps to * that: the handle is taken there, and key frames are asked for there, once * the sender or receiver is gone not at all. + *

+ * The futures of {@link #of(Sender)} and {@link #of(RemoteTrack)} complete + * in any case, failed once the peer or its runtime is closed. A caller that + * stops waiting cancels the future and closes a source that completes it all + * the same, see {@link Source}. */ public final class EncodedFrames { @@ -55,31 +62,54 @@ public interface Access { } /** - * Where a sender's or receiver's encoded frames come from. + * Where a sender's or receiver's encoded frames come from. It owns the + * reference of its handle until {@link #take()} hands it over; closing it + * before drops the reference, closing it after does nothing. Whoever gets + * a source does one or the other, so that no reference is left behind. */ - public static final class Source { + public static final class Source implements AutoCloseable { - private final long frames; + /** The handle, or 0 once taken or released. */ + private final AtomicLong frames; private final Runnable keyFrameRequest; /** * @param frames the handle of {@code NativeApi.encodedFramesOf}, - * whose reference the receiver of this source - * drops. + * whose reference this source takes over. * @param keyFrameRequest asks for a key frame, from any thread. */ public Source(long frames, Runnable keyFrameRequest) { - this.frames = frames; + this.frames = new AtomicLong(frames); this.keyFrameRequest = requireNonNull(keyFrameRequest, "keyFrameRequest"); } /** - * @return the handle of the encoded frames, with a reference to drop. + * Hands over the handle together with its reference, which the caller + * drops from then on. + * + * @return the handle of the encoded frames. + * + * @throws IllegalStateException if it was taken or released before. + */ + public long take() { + long handle = frames.getAndSet(0); + + if (handle == 0) { + throw new IllegalStateException("The encoded frames were taken or released"); + } + + return handle; + } + + /** + * Drops the reference of the handle, unless {@link #take()} handed it + * over. Closing it again does nothing. */ - public long frames() { - return frames; + @Override + public void close() { + NativeApi.releaseEncodedFrames(frames.getAndSet(0)); } /** diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/internal/NativeApi.java b/webrtc/src/main/java/dev/onvoid/webrtc/internal/NativeApi.java index 93b0a576..d6a5089f 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/internal/NativeApi.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/internal/NativeApi.java @@ -128,8 +128,24 @@ public static long encodedFramesOf(RTCRtpReceiver receiver) { return receiverFrames(handleOf(receiver)); } + /** + * Drops the reference a handle of {@link #encodedFramesOf(RTCRtpSender)} + * or {@link #encodedFramesOf(RTCRtpReceiver)} carries, as {@code + * encoded_frames_release()} does, for a handle that never reaches native + * code. The handle must not be used afterwards. + * + * @param frames The handle, or {@code 0}, which is ignored. + */ + public static void releaseEncodedFrames(long frames) { + if (frames != 0) { + releaseFrames(frames); + } + } + private static native long senderFrames(long senderHandle); private static native long receiverFrames(long receiverHandle); + private static native void releaseFrames(long frames); + } diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/EncodedFramesTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/EncodedFramesTests.java new file mode 100644 index 00000000..437638de --- /dev/null +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/EncodedFramesTests.java @@ -0,0 +1,169 @@ +/* + * Copyright 2026 Alex Andres + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.onvoid.webrtc.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.RejectedExecutionException; + +import dev.onvoid.webrtc.client.internal.EncodedFrames; +import dev.onvoid.webrtc.internal.NativeApi; + +import org.junit.jupiter.api.Test; + +/** + * Hands out the encoded frames of senders and remote tracks, as the media + * module's recorder takes them: the futures settle whatever the runtime + * does, and each source's reference is handed over or dropped once. + */ +class EncodedFramesTests extends ClientTestBase { + + @Test + void sourceHandedOverOnce() throws Exception { + Peer peer = rtc.createPeer(manualOptions()); + Sender sender = peer.addTrack(rtc.media().customVideo("video")); + + EncodedFrames.Source taken = await(EncodedFrames.of(sender)); + long frames = taken.take(); + + assertNotEquals(0, frames); + assertThrows(IllegalStateException.class, taken::take); + + // The reference went with the handle, so this drops nothing. + taken.close(); + NativeApi.releaseEncodedFrames(frames); + + EncodedFrames.Source closed = await(EncodedFrames.of(sender)); + closed.close(); + closed.close(); + + assertThrows(IllegalStateException.class, closed::take); + } + + @Test + void senderAfterRuntimeClosed() throws Exception { + Peer peer = rtc.createPeer(manualOptions()); + Sender sender = peer.addTrack(rtc.media().customVideo("video")); + await(EncodedFrames.of(sender)).close(); + + rtc.close(); + + assertClosed(EncodedFrames.of(sender)); + } + + @Test + void remoteTrackAfterRuntimeClosed() throws Exception { + Peer[] peers = pair(); + CompletableFuture received = new CompletableFuture<>(); + + peers[1].onTrack(received::complete); + peers[0].addTrack(rtc.media().customVideo("video")); + + RemoteTrack track = await(received); + await(EncodedFrames.of(track)).close(); + + rtc.close(); + + assertClosed(EncodedFrames.of(track)); + } + + @Test + void rejectedTaskFailsFuture() throws Exception { + // The control pool is full, not shut down, so nothing but the task's + // own fallback fails the future. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = runtime(control); + + try { + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + Sender sender = await(created).addTrack(runtime.media().customVideo("video")); + control.runAll(); + + control.rejecting(true); + CompletableFuture frames = EncodedFrames.of(sender); + + assertClosed(frames); + assertEquals(1, errors.size(), () -> "Errors: " + errors); + assertInstanceOf(RejectedExecutionException.class, errors.get(0).getCause()); + } + finally { + control.rejecting(false); + errors.clear(); + close(runtime, control); + } + } + + @Test + void cancelledBeforeTaken() throws Exception { + // The recorder stopped waiting before the peer got to it, so no + // source is taken for nobody. + ManualExecutor control = new ManualExecutor(); + WebRtc runtime = runtime(control); + + try { + CompletableFuture created = CompletableFuture.supplyAsync(() -> runtime.createPeer(manualOptions())); + control.runUntil(created::isDone, TIMEOUT_SECONDS); + Sender sender = await(created).addTrack(runtime.media().customVideo("video")); + control.runAll(); + + CompletableFuture frames = EncodedFrames.of(sender); + assertEquals(1, control.queued()); + + frames.cancel(false); + control.runAll(); + + assertTrue(frames.isCancelled()); + } + finally { + close(runtime, control); + } + } + + private WebRtc runtime(ManualExecutor control) { + return WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .controlPool(control) + .build(); + } + + private static void close(WebRtc runtime, ManualExecutor control) throws Exception { + CompletableFuture closed = runtime.closeAsync(); + control.runUntil(closed::isDone, TIMEOUT_SECONDS); + await(closed); + } + + private static void assertClosed(CompletableFuture frames) { + // Settled at once, not after a caller's timeout. + assertTrue(frames.isDone()); + + ExecutionException e = assertThrows(ExecutionException.class, frames::get); + assertInstanceOf(PeerClosedException.class, e.getCause()); + } + + private static PeerOptions manualOptions() { + return PeerOptions.builder().negotiation(Negotiation.MANUAL).build(); + } + +}