fix: wait for device switches on close and keep cleared or oversized audio from being sent

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

@@ -19,6 +19,7 @@
 import static java.util.Objects.requireNonNull;
 
 import java.util.ArrayDeque;
+import java.util.Arrays;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -133,6 +134,8 @@ void beforeRelease() {
 	/** Queues what is left of the previous format, filled with silence. */
 	private void flushPartial() {
 		if (partialLength > 0) {
+			// What lies beyond may be audio that clear() dropped.
+			Arrays.fill(partial, partialLength, partial.length, (byte) 0);
 			queue.add(new Chunk(partial, format));
 		}
 

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

@@ -225,8 +225,24 @@ 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.
+	 */
 	private <T> CompletableFuture<T> onMediaExecutor(Supplier<T> action) {
 		CompletableFuture<T> result = new CompletableFuture<>();
+		LocalMediaImpl.Creation operation;
+
+		try {
+			operation = media.begin();
+		}
+		catch (PeerClosedException e) {
+			result.completeExceptionally(e);
+			return result;
+		}
+
 		media.rtc.track(result, null);
 
 		media.executor.execute(() -> {
@@ -237,6 +253,12 @@ private <T> CompletableFuture<T> onMediaExecutor(Supplier<T> action) {
 				result.completeExceptionally(e instanceof WebRtcException
 						? e : new MediaException("Switching the device failed: " + e.getMessage(), e));
 			}
+			finally {
+				operation.close();
+			}
+		}, () -> {
+			operation.close();
+			result.completeExceptionally(new PeerClosedException("The runtime is closed"));
 		});
 
 		return result;

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

@@ -57,7 +57,10 @@ final class LocalMediaImpl implements LocalMedia {
 
 	private final Set<LocalTrackImpl> tracks = Collections.newSetFromMap(new ConcurrentHashMap<>());
 
-	/** Completed as the creations of tracks end, which the runtime waits for when it closes. */
+	/**
+	 * Completed as the creations of tracks and the device switches end, which
+	 * the runtime waits for when it closes.
+	 */
 	private final Set<CompletableFuture<Void>> creations = Collections.newSetFromMap(new ConcurrentHashMap<>());
 
 	// Guarded by cameras.
@@ -421,7 +424,10 @@ private static String newId(String kind) {
 
 
 
-	/** A creation of a track, which the runtime waits for when it closes. */
+	/**
+	 * A creation of a track, or a device switch, which the runtime waits for
+	 * when it closes.
+	 */
 	final class Creation implements AutoCloseable {
 
 		final CompletableFuture<Void> done = new CompletableFuture<>();

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

@@ -16,6 +16,8 @@
 
 package dev.onvoid.webrtc.client;
 
+import dev.onvoid.webrtc.media.audio.CustomAudioSource;
+
 /**
  * The format of 16-bit PCM audio: sample rate and channel count.
  */
@@ -34,7 +36,10 @@ private PcmFormat(int sampleRate, int channels) {
 	/**
 	 * @param sampleRate the sample rate in Hz, a multiple of 100 so that 10 ms
 	 *                   hold a whole number of samples.
-	 * @param channels   the channel count, 1 to 8.
+	 * @param channels   the channel count, 1 to 8. 10 ms of all channels
+	 *                   together may hold at most {@value
+	 *                   CustomAudioSource#MAX_SAMPLES_PER_PUSH} samples, as
+	 *                   48 kHz with 8 channels or 384 kHz with 2 do.
 	 *
 	 * @return the format.
 	 */
@@ -46,6 +51,16 @@ public static PcmFormat of(int sampleRate, int channels) {
 			throw new IllegalArgumentException("The channel count must be between 1 and 8, got " + channels);
 		}
 
+		// WebRTC takes audio in 10 ms chunks of a bounded size. A larger chunk
+		// would be queued all the same and fail only when it is sent.
+		int samples = sampleRate / 100 * channels;
+
+		if (samples > CustomAudioSource.MAX_SAMPLES_PER_PUSH) {
+			throw new IllegalArgumentException("10 ms of " + sampleRate + " Hz with " + channels + " channels hold "
+					+ samples + " samples, more than the " + CustomAudioSource.MAX_SAMPLES_PER_PUSH
+					+ " WebRTC takes at once. Use fewer channels or a lower sample rate");
+		}
+
 		return new PcmFormat(sampleRate, channels);
 	}
 

webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java

@@ -29,6 +29,7 @@
 import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -85,6 +86,71 @@ void pushedAudioArrives() throws Exception {
 		assertEquals(chunk.frames() * chunk.channels() * 2, chunk.samples().length);
 	}
 
+	@Test
+	void clearedAudioNotResent() throws Exception {
+		// A format change sends the chunk begun before it, filled with
+		// silence, not with the audio that clear() dropped from it.
+		ManualTimer timer = new ManualTimer();
+		WebRtc runtime = WebRtc.builder()
+				.audio(AudioMode.HEADLESS)
+				.errorHandler(errors::add)
+				.timer(timer)
+				.build();
+		LocalMediaImpl media = (LocalMediaImpl) runtime.media();
+		List<byte[]> sent = new ArrayList<>();
+		CustomAudioImpl voice;
+
+		try (LocalMediaImpl.Creation creation = media.begin()) {
+			CustomAudioSource source = new CustomAudioSource() {
+
+				@Override
+				public void pushAudio(byte[] audioData, int bitsPerSample, int sampleRate, int channels,
+						int frameCount) {
+					sent.add(audioData.clone());
+				}
+			};
+
+			voice = media.register(new CustomAudioImpl(media, runtime.factory.createAudioTrack("voice", source),
+					source));
+		}
+
+		try {
+			// 160 bytes in 10 ms.
+			PcmFormat format = PcmFormat.of(8000, 1);
+			byte[] dropped = new byte[120];
+			byte[] kept = new byte[20];
+			Arrays.fill(dropped, (byte) 1);
+			Arrays.fill(kept, (byte) 2);
+
+			voice.push(dropped, format);
+			voice.clear();
+			voice.push(kept, format);
+			voice.push(new byte[2], PcmFormat.of(16000, 1));
+
+			timer.fire(Duration.ofMillis(10));
+
+			assertEquals(1, sent.size());
+			assertEquals(160, sent.get(0).length);
+
+			for (int i = 0; i < 160; i++) {
+				assertEquals(i < 20 ? 2 : 0, sent.get(0)[i], "Byte " + i);
+			}
+		}
+		finally {
+			runtime.close();
+		}
+	}
+
+	@Test
+	void oversizedFormatRejected() {
+		// 10 ms of these hold 30720 samples, more than WebRTC takes at once.
+		assertThrows(IllegalArgumentException.class, () -> PcmFormat.of(384000, 8));
+
+		// The largest that fit, 7680 and 3840 samples.
+		PcmFormat.of(384000, 2);
+		PcmFormat.of(48000, 8);
+	}
+
 	@Test
 	void pushedVideoArrives() throws Exception {
 		Peer[] peers = pair();

webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import dev.onvoid.webrtc.media.audio.AudioDevice;
 import dev.onvoid.webrtc.media.video.CustomVideoSource;
 
 import org.junit.jupiter.api.Test;
@@ -319,6 +320,46 @@ void beforeRelease() {
 		}
 	}
 
+	@Test
+	void closeWhileSwitchingDevice() throws Exception {
+		// A speaker switch is queued when the runtime begins to close. The
+		// runtime waits for it before it frees the audio module it uses.
+		ManualExecutor control = new ManualExecutor();
+		WebRtc runtime = runtime(control);
+		LocalMediaImpl media = (LocalMediaImpl) runtime.media();
+		AtomicBoolean poolStopped = new AtomicBoolean(true);
+
+		CompletableFuture<Void> switched = media.devices().useSpeaker(speaker());
+		switched.whenComplete((value, error) -> poolStopped.set(runtime.controlPool.isShutdown()));
+
+		// Named among what the runtime waits for.
+		assertEquals(1, media.closeAll().size());
+
+		CompletableFuture<Void> closed = runtime.closeAsync();
+		control.runUntil(closed::isDone, TIMEOUT_SECONDS);
+		await(closed);
+
+		// The switch ran while the runtime still had its threads, and failed
+		// on its own: a headless runtime plays to no speakers.
+		assertFalse(poolStopped.get());
+		ExecutionException e = assertThrows(ExecutionException.class, () -> await(switched));
+		assertInstanceOf(ConfigurationException.class, e.getCause());
+	}
+
+	@Test
+	void switchDeviceAfterClose() throws Exception {
+		Devices devices = rtc.media().devices();
+		CompletableFuture<Void> closed = rtc.closeAsync();
+
+		// Failed at once, without reaching the module the runtime frees.
+		CompletableFuture<Void> switched = devices.useSpeaker(speaker());
+
+		assertTrue(switched.isDone());
+		ExecutionException e = assertThrows(ExecutionException.class, () -> await(switched));
+		assertInstanceOf(PeerClosedException.class, e.getCause());
+		await(closed);
+	}
+
 	private WebRtc runtime(ManualExecutor control) {
 		return WebRtc.builder()
 				.audio(AudioMode.HEADLESS)
@@ -327,6 +368,10 @@ private WebRtc runtime(ManualExecutor control) {
 				.build();
 	}
 
+	private static AudioDevice speaker() {
+		return new AudioDevice("Test speaker", "test-speaker") { };
+	}
+
 	private static PeerOptions manualOptions() {
 		return PeerOptions.builder().negotiation(Negotiation.MANUAL).build();
 	}