fix: keep local track releases independent of handlers and cancelled futures

docs/guide/client-api.md

@@ -204,7 +204,7 @@ Every method of the client API may be called from any thread, including from ins
 | Your calls into the API | your thread | They don't block. The exceptions are `WebRtc.createPeer`, which waits briefly for the connection, and `WebRtc.close()`, which waits up to the close timeout. |
 | Peer, track and channel handlers, future callbacks, `Signaling.send` | the peer's event thread | Yes, but that delays only the events of this peer. They must not wait for a future of this API, see below. |
 | `onFrame` and `onAudio` handlers | a thread of each subscription | Better not. A slow video handler gets the newest frame next, and the frames in between are dropped; a slow audio handler loses the oldest audio beyond its buffer. |
-| `LocalTrack.onEnded` and `Devices.onChange` handlers | a background thread of the runtime | Briefly. |
+| `LocalTrack.onEnded` and `Devices.onChange` handlers, callbacks of the futures of local tracks and devices | the event thread of the runtime's local media | Yes, but that delays the other events of local media. They must not wait for a future of this API either. A track is released before its `onEnded` handlers run, so they never hold up a release. |
 | WebRTC's own callbacks | WebRTC's threads | Not your code. The library only copies what they report and queues it. |
 
 The handlers of one peer never run at the same time, and they run in the order of the events. The handlers of different peers may run at the same time. With the default executor, the event thread is named after its peer, such as `webrtc-client-events-alice`, which shows up in thread dumps.
@@ -329,6 +329,8 @@ peer.onChannel(channel -> {
 
 A future that is already done may be read anywhere, and any thread that is not an event thread, such as `main`, may wait.
 
+Each call returns a future of its own, also of `closeAsync()`, `whenClosed()`, `whenConnected()` and `whenOpen()`. Cancelling one only stops waiting: the close goes on, and the futures of other callers, and of the runtime's own close, are not affected.
+
 ::: warning
 On Java 8, only the futures the API returns refuse to be waited on. Futures derived from them with `thenApply` and the like do not, so waiting for one of those in a handler hangs that peer. From Java 9 on, the derived futures refuse too.
 

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

@@ -46,6 +46,21 @@ static <T> ClientFuture<T> completed(T value) {
 		return future;
 	}
 
+	/**
+	 * Hands out a future of its own for a future shared by many callers: a
+	 * caller that cancels or completes its copy changes nothing for the
+	 * others, nor for what waits for the shared one.
+	 *
+	 * @param shared the future the copy completes with, on the same thread.
+	 *
+	 * @return a new future.
+	 */
+	static <T> ClientFuture<T> copyOf(CompletableFuture<T> shared) {
+		ClientFuture<T> copy = new ClientFuture<>();
+		shared.whenComplete((value, error) -> PeerImpl.complete(copy, value, error));
+		return copy;
+	}
+
 	@Override
 	public T get() throws InterruptedException, ExecutionException {
 		checkBlocking();

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

@@ -157,7 +157,7 @@ public ChannelState state() {
 
 	@Override
 	public CompletableFuture<DataChannel> whenOpen() {
-		return opened;
+		return ClientFuture.copyOf(opened);
 	}
 
 	@Override

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

@@ -85,7 +85,8 @@ public interface Devices {
 	/**
 	 * Calls the handler after a device was connected or disconnected.
 	 *
-	 * @param handler called on a background thread with the new snapshot.
+	 * @param handler called on an event thread of the runtime with the new
+	 *                snapshot.
 	 *
 	 * @return the subscription that removes the handler.
 	 */

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

@@ -164,7 +164,7 @@ void dispose() {
 	private void changed() {
 		media.executor.execute(() -> {
 			refresh();
-			handlers.accept(this);
+			media.events.execute(() -> handlers.accept(this));
 		});
 	}
 
@@ -247,7 +247,9 @@ private static List<DesktopSource> desktopSources(DesktopCapturer capturer) {
 	 * Runs a switch, which uses the audio module, on the media executor. It
 	 * is registered like the creation of a track, so that a runtime that
 	 * begins to close either waits for it before it frees the module, or is
-	 * seen closing here.
+	 * seen closing here. The future handed out completes on the event
+	 * executor, so that the application's callbacks never hold up the media
+	 * executor.
 	 */
 	private <T> CompletableFuture<T> onMediaExecutor(Supplier<T> action) {
 		CompletableFuture<T> result = new CompletableFuture<>();
@@ -279,6 +281,6 @@ private <T> CompletableFuture<T> onMediaExecutor(Supplier<T> action) {
 			result.completeExceptionally(new PeerClosedException("The runtime is closed"));
 		});
 
-		return result;
+		return media.toPublic(result);
 	}
 }

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

@@ -113,7 +113,7 @@ public CompletableFuture<Void> closeAsync() {
 					.whenComplete((ignored, error) -> closed.complete(null));
 		}
 
-		return closed;
+		return ClientFuture.copyOf(closed);
 	}
 
 	/**

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

@@ -75,7 +75,9 @@ public CompletableFuture<Void> switchCamera(VideoDevice device) {
 			}
 		});
 
-		return switched;
+		// Completed on the event executor, so that the application's
+		// callbacks never hold up the media executor.
+		return media.toPublic(switched);
 	}
 
 	private void switchTo(VideoDevice device) {

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

@@ -50,9 +50,16 @@ final class LocalMediaImpl implements LocalMedia {
 
 	final WebRtc rtc;
 
-	/** Releases tracks and runs their ended handlers, off WebRTC's and the application's threads. */
+	/** Releases tracks and switches devices, off WebRTC's and the application's threads. */
 	final SerialExecutor executor;
 
+	/**
+	 * Runs the ended and device handlers and completes the futures handed
+	 * out, as a peer's event executor does, so that no application code holds
+	 * up a release.
+	 */
+	final SerialExecutor events;
+
 	private final AudioDeviceModuleBase audioModule;
 
 	private final Set<LocalTrackImpl> tracks = Collections.newSetFromMap(new ConcurrentHashMap<>());
@@ -76,6 +83,8 @@ final class LocalMediaImpl implements LocalMedia {
 		this.rtc = rtc;
 		this.audioModule = audioModule;
 		this.executor = new SerialExecutor(rtc.controlPool, rtc.controlWork, rtc::reportError, false, null);
+		this.events = new SerialExecutor(rtc.eventExecutor, rtc.eventWork, rtc::reportError, true,
+				rtc.ownsEventExecutor ? "webrtc-client-events-media" : null);
 	}
 
 	@Override
@@ -310,6 +319,29 @@ List<CompletableFuture<Void>> closeAll() {
 		return closing;
 	}
 
+	/**
+	 * Hands a future out to the application: it completes on the event
+	 * executor, after the handlers queued before, and cancelling it leaves the
+	 * internal one alone.
+	 */
+	<T> CompletableFuture<T> toPublic(CompletableFuture<T> internal) {
+		ClientFuture<T> result = new ClientFuture<>();
+
+		internal.whenComplete((value, error) -> {
+			Runnable deliver = () -> PeerImpl.complete(result, value, error);
+
+			// Completed in any case: should the event executor reject the
+			// task, on the thread that found it rejected.
+			events.execute(deliver, deliver);
+		});
+
+		// Tracked after, so that a track released before the runtime stopped
+		// its threads still reports that, rather than the runtime's close.
+		rtc.track(result, events);
+
+		return result;
+	}
+
 	synchronized void dispose() {
 		if (devices != null) {
 			devices.dispose();

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

@@ -48,10 +48,12 @@ public interface LocalTrack extends AutoCloseable {
 	void setEnabled(boolean enabled);
 
 	/**
-	 * Calls the handler once the track ended. A handler added after that is
-	 * called at once.
+	 * Calls the handler once the track ended and its native track is
+	 * released. A handler added after that is called at once.
 	 *
-	 * @param handler called on a background thread.
+	 * @param handler called on an event thread of the runtime, where waiting
+	 *                for a future of this API that is not done throws a
+	 *                {@link BlockingCallException}.
 	 *
 	 * @return the subscription that removes the handler.
 	 */
@@ -71,6 +73,9 @@ public interface LocalTrack extends AutoCloseable {
 	 * LocalMedia#videoTrack} or {@link LocalMedia#audioTrack}, may be closed;
 	 * or fails with a {@link WebRtcException} when the track or its source
 	 * could not be released, which then fails the runtime's close as well.
+	 * It completes on an event thread, after the ended handlers ran. Each
+	 * call returns a future of its own: cancelling it stops waiting, not the
+	 * close.
 	 */
 	CompletableFuture<Void> closeAsync();
 

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

@@ -32,7 +32,9 @@
  * Calls on the native track and source happen under this object's lock and
  * only while the track has not ended, so that closing never races with them.
  * Closing first makes every sender let go of the track, each on its peer's
- * control executor, and only then releases the native track and source.
+ * control executor, and only then releases the native track and source. The
+ * ended handlers and the futures handed out follow on the media's event
+ * executor, so that neither holds up a release nor the runtime's close.
  */
 abstract class LocalTrackImpl implements LocalTrack {
 
@@ -50,7 +52,12 @@ abstract class LocalTrackImpl implements LocalTrack {
 	/** Guarded by this. */
 	private Runnable releaseSource;
 
-	private final CompletableFuture<Void> closed = new CompletableFuture<>();
+	/**
+	 * Completed on the media executor once the native track is released,
+	 * before the handlers hear of it. Never handed out, so that no caller can
+	 * cancel what the runtime waits for.
+	 */
+	private final CompletableFuture<Void> closedInternal = new CompletableFuture<>();
 
 	private volatile boolean enabled = true;
 
@@ -117,7 +124,7 @@ public Subscription onEnded(Runnable handler) {
 			}
 		}
 
-		media.executor.execute(handler);
+		media.events.execute(handler);
 
 		return () -> { };
 	}
@@ -129,7 +136,7 @@ public synchronized boolean isEnded() {
 
 	@Override
 	public CompletableFuture<Void> closeAsync() {
-		return closeInternal();
+		return media.toPublic(closeInternal());
 	}
 
 	@Override
@@ -205,12 +212,18 @@ boolean isLive() {
 		return !ended;
 	}
 
+	/**
+	 * Starts closing, once.
+	 *
+	 * @return a future that completes on the media executor once the native
+	 * track is released, before the handlers hear of it.
+	 */
 	CompletableFuture<Void> closeInternal() {
 		List<SenderImpl> using;
 
 		synchronized (this) {
 			if (ended) {
-				return closed;
+				return closedInternal;
 			}
 
 			ended = true;
@@ -227,7 +240,7 @@ CompletableFuture<Void> closeInternal() {
 		CompletableFuture.allOf(detached.toArray(new CompletableFuture[0]))
 				.whenComplete((ignored, error) -> media.executor.execute(this::release));
 
-		return closed;
+		return closedInternal;
 	}
 
 	/**
@@ -270,20 +283,25 @@ private void release() {
 			endedHandlers.clear();
 		}
 
-		for (Runnable handler : handlers) {
-			try {
-				handler.run();
-			}
-			catch (Throwable e) {
-				rtc.reportError(e);
+		// Queued before the release completes, so that the futures handed out
+		// complete after the handlers ran, and the runtime cannot stop the
+		// event executor in between.
+		media.events.execute(() -> {
+			for (Runnable handler : handlers) {
+				try {
+					handler.run();
+				}
+				catch (Throwable e) {
+					rtc.reportError(e);
+				}
 			}
-		}
+		});
 
 		if (failure == null) {
-			closed.complete(null);
+			closedInternal.complete(null);
 		}
 		else {
-			closed.completeExceptionally(failure);
+			closedInternal.completeExceptionally(failure);
 		}
 	}
 

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

@@ -401,12 +401,12 @@ public Subscription onError(Consumer<WebRtcException> handler) {
 
 	@Override
 	public CompletableFuture<Peer> whenConnected() {
-		return connected;
+		return ClientFuture.copyOf(connected);
 	}
 
 	@Override
 	public CompletableFuture<Void> whenClosed() {
-		return closedFuture;
+		return ClientFuture.copyOf(closedFuture);
 	}
 
 	@Override
@@ -522,7 +522,7 @@ public <T> CompletableFuture<T> withNative(Function<RTCPeerConnection, T> action
 	public CompletableFuture<Void> closeAsync() {
 		closeInternal();
 
-		return closedFuture;
+		return ClientFuture.copyOf(closedFuture);
 	}
 
 	@Override

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

@@ -135,6 +135,7 @@ public final class WebRtc implements AutoCloseable {
 
 	private final AtomicBoolean closing = new AtomicBoolean();
 
+	/** Never handed out, so that no caller can cancel what the close waits for. */
 	private final CompletableFuture<Void> closed = new CompletableFuture<>();
 
 	/** The futures tracked that are not done, with what fails each. */
@@ -319,16 +320,17 @@ public LoadStats loadStats() {
 	 * that names the cause, as it does when disposing fails.
 	 *
 	 * @return a future that completes once all is closed, or fails when not
-	 * everything could be released.
+	 * everything could be released. Each call returns a future of its own:
+	 * cancelling it stops waiting, not the close.
 	 */
 	public CompletableFuture<Void> closeAsync() {
-		if (closing.compareAndSet(false, true)) {
-			Thread thread = new Thread(this::shutdown, "webrtc-client-close");
-			thread.setDaemon(true);
-			thread.start();
-		}
+		startClose();
+
+		// Not on an event thread, so that one may wait for it.
+		CompletableFuture<Void> result = new CompletableFuture<>();
+		closed.whenComplete((ignored, error) -> PeerImpl.complete(result, null, error));
 
-		return closed;
+		return result;
 	}
 
 	/**
@@ -343,10 +345,10 @@ public CompletableFuture<Void> closeAsync() {
 	 */
 	@Override
 	public void close() {
-		CompletableFuture<Void> future = closeAsync();
+		startClose();
 
 		try {
-			future.get(Timeouts.toMillis(closeTimeout), TimeUnit.MILLISECONDS);
+			closed.get(Timeouts.toMillis(closeTimeout), TimeUnit.MILLISECONDS);
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
@@ -359,6 +361,14 @@ public void close() {
 		}
 	}
 
+	private void startClose() {
+		if (closing.compareAndSet(false, true)) {
+			Thread thread = new Thread(this::shutdown, "webrtc-client-close");
+			thread.setDaemon(true);
+			thread.start();
+		}
+	}
+
 	/**
 	 * Registers a peer before the runtime is checked again: a runtime that
 	 * begins to close either finds the peer among those it closes and waits