From 4fe49de31782f08bb1f5d63e442fbda1b8284753 Mon Sep 17 00:00:00 2001 From: Alex Andres Date: Sat, 24 Oct 2026 02:51:04 +0200 Subject: [PATCH] fix: wait for device switches on close and keep cleared or oversized audio from being sent Device switches register with the runtime like track creations, so closing waits for them before freeing the audio module and rejects new ones. A chunk cut short by a format change is padded with silence instead of stale bytes left over from clear(), and PcmFormat.of rejects formats whose 10 ms chunk exceeds what CustomAudioSource accepts. --- .../onvoid/webrtc/client/CustomAudioImpl.java | 3 + .../dev/onvoid/webrtc/client/DevicesImpl.java | 22 +++++++ .../onvoid/webrtc/client/LocalMediaImpl.java | 10 ++- .../dev/onvoid/webrtc/client/PcmFormat.java | 17 ++++- .../dev/onvoid/webrtc/client/MediaTests.java | 66 +++++++++++++++++++ .../webrtc/client/RuntimeCloseTests.java | 45 +++++++++++++ 6 files changed, 160 insertions(+), 3 deletions(-) diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java index d24ee024..a9d09633 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java +++ b/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)); } diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java index 04de2e14..2da93720 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/DevicesImpl.java @@ -225,8 +225,24 @@ private static List 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 CompletableFuture onMediaExecutor(Supplier action) { CompletableFuture 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 CompletableFuture onMediaExecutor(Supplier 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; 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 58ad24e0..e55802fe 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/LocalMediaImpl.java @@ -57,7 +57,10 @@ final class LocalMediaImpl implements LocalMedia { private final Set 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> 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 done = new CompletableFuture<>(); diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/PcmFormat.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/PcmFormat.java index c352d474..361153a6 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/PcmFormat.java +++ b/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); } diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java index 70125e8c..d44de35d 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java +++ b/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 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(); diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java index cd6a3b66..b5efc66d 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/RuntimeCloseTests.java +++ b/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 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 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 closed = rtc.closeAsync(); + + // Failed at once, without reaching the module the runtime frees. + CompletableFuture 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(); }