fix: settle and release recorder sources of the client API

webrtc-java-media/src/main/java/dev/onvoid/webrtc/media/recorder/MediaRecorder.java

@@ -316,30 +316,42 @@ 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);
 	}
 
 	/**
 	 * Waits for the client API's peer to hand over the encoded frames. Its
 	 * control thread never takes this recorder's lock, so waiting under it
 	 * cannot deadlock.
 	 */
-	private static EncodedFrames.Source sourceOf(CompletableFuture<EncodedFrames.Source> future) {
+	static EncodedFrames.Source sourceOf(CompletableFuture<EncodedFrames.Source> 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);
 		}
 		catch (ExecutionException e) {
 			throw new IllegalArgumentException(e.getCause().getMessage(), e.getCause());
 		}
 		catch (TimeoutException e) {
+			giveUp(future);
 			throw new IllegalStateException("The peer did not hand over the track in time", e);
 		}
 	}
 
+	/**
+	 * Stops waiting for a source. The peer closes one that comes after the
+	 * cancellation; this closes one that came before it, since no recorder
+	 * takes it over.
+	 */
+	static void giveUp(CompletableFuture<EncodedFrames.Source> 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");

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<EncodedFrames.Source> 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<EncodedFrames.Source> 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);

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

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<void *>(frames));
+}

webrtc/src/main/java/dev/onvoid/webrtc/client/PeerImpl.java

@@ -576,6 +576,44 @@ <T> CompletableFuture<T> toPublic(CompletableFuture<T> 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.Source> encodedFrames(Supplier<EncodedFrames.Source> take) {
+		CompletableFuture<EncodedFrames.Source> 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()) {

webrtc/src/main/java/dev/onvoid/webrtc/client/RemoteTrackImpl.java

@@ -157,32 +157,23 @@ void onControl(Runnable change) {
 	 * too.
 	 */
 	CompletableFuture<EncodedFrames.Source> encodedFrames() {
-		CompletableFuture<EncodedFrames.Source> 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. */

webrtc/src/main/java/dev/onvoid/webrtc/client/SenderImpl.java

@@ -314,30 +314,21 @@ CompletableFuture<Void> trackReplaced(LocalTrackImpl replacedIn) {
 	 * peer's control executor, where key frames are asked for too.
 	 */
 	CompletableFuture<EncodedFrames.Source> encodedFrames() {
-		CompletableFuture<EncodedFrames.Source> 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;
 	}
 
 	/**

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.
+ * <p>
+ * 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));
 		}
 
 		/**

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);
+
 }