diff --git a/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/AudioPlaybackSample.java b/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/AudioPlaybackSample.java index 9f7b43e8fec7..0941c8d675ff 100644 --- a/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/AudioPlaybackSample.java +++ b/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/AudioPlaybackSample.java @@ -13,28 +13,32 @@ import com.azure.ai.voicelive.models.OpenAIVoiceName; import com.azure.ai.voicelive.models.OutputAudioFormat; import com.azure.ai.voicelive.models.ServerEventType; +import com.azure.ai.voicelive.models.SessionResponse; +import com.azure.ai.voicelive.models.SessionResponseStatus; import com.azure.ai.voicelive.models.SessionServerEvent; import com.azure.ai.voicelive.models.SessionUpdateError; import com.azure.ai.voicelive.models.SessionUpdateResponseAudioDelta; +import com.azure.ai.voicelive.models.SessionUpdateResponseAudioDone; +import com.azure.ai.voicelive.models.SessionUpdateResponseDone; import com.azure.ai.voicelive.models.UserMessageItem; import com.azure.ai.voicelive.models.VoiceLiveSessionOptions; import com.azure.core.util.BinaryData; import com.azure.identity.DefaultAzureCredentialBuilder; import reactor.core.publisher.Mono; -import java.util.Collections; - import javax.sound.sampled.AudioFormat; import javax.sound.sampled.AudioSystem; import javax.sound.sampled.DataLine; import javax.sound.sampled.LineUnavailableException; import javax.sound.sampled.SourceDataLine; +import java.time.Duration; import java.util.Arrays; +import java.util.Collections; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; /** @@ -90,6 +94,7 @@ public final class AudioPlaybackSample { private static final int CHANNELS = 1; // Mono private static final int SAMPLE_SIZE_BITS = 16; // 16-bit PCM private static final int CHUNK_SIZE = 1200; // 50ms chunks + private static final int AUDIO_QUEUE_CAPACITY = 1000; private static final long COMPLETION_TIMEOUT_SECONDS = 60; /** @@ -98,7 +103,6 @@ public final class AudioPlaybackSample { * @param args Unused command line arguments */ public static void main(String[] args) { - // Get endpoint from environment variable String endpoint = System.getenv("AZURE_VOICELIVE_ENDPOINT"); if (endpoint == null) { @@ -106,13 +110,11 @@ public static void main(String[] args) { return; } - // Check if speaker is available if (!checkSpeakerAvailable()) { System.err.println("No compatible speaker found"); return; } - // Create the VoiceLive client using DefaultAzureCredential (Entra ID). VoiceLiveAsyncClient client = new VoiceLiveClientBuilder() .endpoint(endpoint) .credential(new DefaultAzureCredentialBuilder().build()) @@ -120,86 +122,62 @@ public static void main(String[] args) { System.out.println("Starting audio playback sample..."); - // Configure session options + PlaybackController playback; + try { + playback = startPlayback(); + } catch (LineUnavailableException e) { + System.err.println("Failed to start speaker: " + e.getMessage()); + return; + } + + Mono sample = Mono.usingWhen( + client.startSession("gpt-realtime", null), + session -> { + Mono responseCompletion = configureSession(session) + .then(sendPrompt(session)) + .thenMany(session.receiveEvents().concatMap(event -> handleEvent(event, playback))) + .filter(Boolean::booleanValue) + .next() + .switchIfEmpty(Mono.error(new IllegalStateException( + "Event stream completed before a successful response.done event"))) + .then(); + Mono playbackFailure = playback.completion().then(Mono.never()); + return Mono.firstWithSignal(responseCompletion, playbackFailure); + }, + VoiceLiveSessionAsyncClient::closeAsync) + .timeout(Duration.ofSeconds(COMPLETION_TIMEOUT_SECONDS)) + .doOnError(playback::abort) + .doFinally(signalType -> playback.close()); + + try { + sample.block(); + System.out.println("\nSample completed - all queued audio was played and drained"); + } catch (Exception error) { + System.err.println("Audio playback sample failed: " + rootMessage(error)); + } + } + + private static Mono configureSession(VoiceLiveSessionAsyncClient session) { VoiceLiveSessionOptions sessionOptions = new VoiceLiveSessionOptions() .setInstructions("You are a helpful assistant. Respond to user messages with clear, friendly audio.") - // Voice options: - // - OpenAI: new OpenAIVoice(OpenAIVoiceName.ALLOY) - use OpenAIVoiceName enum - // - Azure: AzureStandardVoice, AzureCustomVoice, AzurePersonalVoice (all extend AzureVoice) .setVoice(BinaryData.fromObject(new OpenAIVoice(OpenAIVoiceName.ALLOY))) .setModalities(Arrays.asList(InteractionModality.TEXT, InteractionModality.AUDIO)) .setInputAudioFormat(InputAudioFormat.PCM16) .setOutputAudioFormat(OutputAudioFormat.PCM16) .setInputAudioSamplingRate(SAMPLE_RATE); - // Audio playback components - final BlockingQueue audioQueue = new LinkedBlockingQueue<>(1000); - final AtomicBoolean isPlaying = new AtomicBoolean(false); - final AtomicReference speakerRef = new AtomicReference<>(); - final AtomicReference playbackThreadRef = new AtomicReference<>(); - - // Latch keeps main alive until the response completes (or an error occurs). - final CountDownLatch completionLatch = new CountDownLatch(1); - - // Open a WebSocket session against the realtime model. - client.startSession("gpt-realtime", null) - // Configure the session (voice, modalities, audio formats, instructions). - .flatMap(session -> { - ClientEventSessionUpdate updateEvent = new ClientEventSessionUpdate(sessionOptions); - return session.sendEvent(updateEvent).thenReturn(session); - }) - // Open the speaker line and start the playback worker thread before - // any audio deltas arrive, so chunks can be played as soon as they stream in. - .flatMap(session -> { - startPlayback(audioQueue, isPlaying, speakerRef, playbackThreadRef); - return Mono.just(session); - }) - // Send a user message that prompts the model to produce a spoken reply. - .flatMap(session -> { - InputTextContentPart textContent = new InputTextContentPart( - "Please say 'Hello! This is a test of the audio playback system.' in a friendly voice."); - UserMessageItem messageItem = new UserMessageItem(Collections.singletonList(textContent)); - ClientEventConversationItemCreate createEvent = new ClientEventConversationItemCreate() - .setItem(messageItem); - return session.sendEvent(createEvent).thenReturn(session); - }) - // Ask the model to start generating a response for the queued message. - .flatMap(session -> { - ClientEventResponseCreate responseEvent = new ClientEventResponseCreate(); - return session.sendEvent(responseEvent).thenReturn(session); - }) - // Subscribe to the server event stream (session.created, audio deltas, etc.). - .flatMapMany(session -> session.receiveEvents()) - .subscribe( - // onNext: route each server event (audio chunks go to the playback queue). - event -> handleEvent(event, audioQueue, completionLatch), - // onError: log and release main so it can clean up and exit. - error -> { - System.err.println("Error: " + error.getMessage()); - completionLatch.countDown(); - }, - // onComplete: stream ended cleanly; release main. - completionLatch::countDown - ); + return session.sendEvent(new ClientEventSessionUpdate(sessionOptions)); + } - try { - if (!completionLatch.await(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { - System.err.println("Timed out waiting for audio response to complete."); - } else { - System.out.println("\nāœ“ Sample completed - audio playback demonstrated"); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } finally { - stopPlayback(audioQueue, isPlaying, speakerRef, playbackThreadRef); - } + private static Mono sendPrompt(VoiceLiveSessionAsyncClient session) { + InputTextContentPart textContent = new InputTextContentPart( + "Please say 'Hello! This is a test of the audio playback system.' in a friendly voice."); + UserMessageItem messageItem = new UserMessageItem(Collections.singletonList(textContent)); + ClientEventConversationItemCreate createEvent = new ClientEventConversationItemCreate().setItem(messageItem); + + return session.sendEvent(createEvent).then(session.sendEvent(new ClientEventResponseCreate())); } - /** - * Check if a compatible speaker is available. - * - * @return true if speaker is available, false otherwise - */ private static boolean checkSpeakerAvailable() { try { AudioFormat format = new AudioFormat(SAMPLE_RATE, SAMPLE_SIZE_BITS, CHANNELS, true, false); @@ -210,143 +188,329 @@ private static boolean checkSpeakerAvailable() { } } + private static PlaybackController startPlayback() throws LineUnavailableException { + AudioFormat format = new AudioFormat( + AudioFormat.Encoding.PCM_SIGNED, + SAMPLE_RATE, + SAMPLE_SIZE_BITS, + CHANNELS, + CHANNELS * SAMPLE_SIZE_BITS / 8, + SAMPLE_RATE, + false); + + DataLine.Info info = new DataLine.Info(SourceDataLine.class, format); + SourceDataLine speaker = (SourceDataLine) AudioSystem.getLine(info); + speaker.open(format, CHUNK_SIZE * 4); + speaker.start(); + + PlaybackController playback = new PlaybackController(new SourceDataLineOutput(speaker), AUDIO_QUEUE_CAPACITY); + playback.start(); + System.out.println("Audio playback started"); + return playback; + } + /** - * Start audio playback system. - * - * @param audioQueue Queue containing audio data to play - * @param isPlaying Flag to control playback loop - * @param speakerRef Reference to store the speaker line - * @param playbackThreadRef Reference to store the playback thread + * Handles one server event and reports whether the successful terminal response was reached. */ - private static void startPlayback(BlockingQueue audioQueue, AtomicBoolean isPlaying, - AtomicReference speakerRef, AtomicReference playbackThreadRef) { - try { - AudioFormat format = new AudioFormat( - AudioFormat.Encoding.PCM_SIGNED, - SAMPLE_RATE, - SAMPLE_SIZE_BITS, - CHANNELS, - CHANNELS * SAMPLE_SIZE_BITS / 8, - SAMPLE_RATE, - false - ); + static Mono handleEvent(SessionServerEvent event, PlaybackController playback) { + ServerEventType eventType = event.getType(); - DataLine.Info info = new DataLine.Info(SourceDataLine.class, format); - SourceDataLine speaker = (SourceDataLine) AudioSystem.getLine(info); - speaker.open(format, CHUNK_SIZE * 4); - speaker.start(); + if (eventType == ServerEventType.SESSION_CREATED) { + System.out.println("Session created"); + } else if (eventType == ServerEventType.SESSION_UPDATED) { + System.out.println("Session updated - ready to receive audio"); + } else if (event instanceof SessionUpdateResponseAudioDelta) { + byte[] audioData = ((SessionUpdateResponseAudioDelta) event).getDelta(); + if (audioData != null && audioData.length > 0 && !playback.queueAudio(audioData)) { + System.err.println("Warning: audio queue full, dropping chunk of " + audioData.length + " bytes"); + } + } else if (event instanceof SessionUpdateResponseAudioDone) { + SessionUpdateResponseAudioDone audioDone = (SessionUpdateResponseAudioDone) event; + System.out.println("Audio response complete: responseId=" + audioDone.getResponseId() + + ", itemId=" + audioDone.getItemId() + + ", outputIndex=" + audioDone.getOutputIndex() + + ", contentIndex=" + audioDone.getContentIndex() + + ", acceptedBytes=" + playback.getAcceptedAudioBytes() + + ", droppedChunks=" + playback.getDroppedAudioChunks()); + } else if (event instanceof SessionUpdateResponseDone) { + SessionResponse response = ((SessionUpdateResponseDone) event).getResponse(); + String responseId = response == null ? null : response.getId(); + SessionResponseStatus status = response == null ? null : response.getStatus(); + System.out.println("Response complete: responseId=" + responseId + ", status=" + status); - speakerRef.set(speaker); - isPlaying.set(true); + try { + validateCompletedResponse(responseId, status, playback.getAcceptedAudioBytes()); + } catch (IllegalStateException error) { + return Mono.error(error); + } - System.out.println("šŸ”Š Audio playback started"); + return playback.finishGracefully().thenReturn(true); + } else if (event instanceof SessionUpdateError) { + SessionUpdateError errorEvent = (SessionUpdateError) event; + String message = errorEvent.getError() == null ? "Unknown VoiceLive error" : errorEvent.getError().getMessage(); + return Mono.error(new IllegalStateException(message)); + } - // Start playback thread - Thread playbackThread = new Thread(() -> { - while (isPlaying.get()) { - try { - byte[] audioData = audioQueue.take(); // Blocking wait + return Mono.just(false); + } - if (audioData.length == 0) { - // Shutdown signal - break; - } + static void validateCompletedResponse(String responseId, SessionResponseStatus status, long acceptedAudioBytes) { + if (!SessionResponseStatus.COMPLETED.equals(status)) { + throw new IllegalStateException( + "Response " + responseId + " ended with status " + status + " instead of completed"); + } + if (acceptedAudioBytes == 0) { + throw new IllegalStateException("Response " + responseId + " completed without any playable audio"); + } + } - // Play the audio - if (speaker.isOpen()) { - speaker.write(audioData, 0, audioData.length); - } + private static String rootMessage(Throwable error) { + Throwable current = error; + while (current.getCause() != null) { + current = current.getCause(); + } + return current.getMessage() == null ? current.getClass().getSimpleName() : current.getMessage(); + } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } catch (Exception e) { - System.err.println("Error in audio playback: " + e.getMessage()); - } - } - }, "AudioPlayback"); - playbackThread.setDaemon(true); - playbackThreadRef.set(playbackThread); - playbackThread.start(); + interface AudioOutput { + boolean isOpen(); - } catch (LineUnavailableException e) { - System.err.println("Failed to start speaker: " + e.getMessage()); - } + int write(byte[] audioData, int offset, int length); + + void drain(); + + void stop(); + + void flush(); + + void close(); } - /** - * Stop audio playback. - * - * @param audioQueue Queue containing audio data - * @param isPlaying Flag to control playback loop - * @param speakerRef Reference to the speaker line to close - * @param playbackThreadRef Reference to the playback thread - */ - private static void stopPlayback(BlockingQueue audioQueue, AtomicBoolean isPlaying, - AtomicReference speakerRef, AtomicReference playbackThreadRef) { - isPlaying.set(false); - audioQueue.offer(new byte[0]); // Shutdown signal + private static final class SourceDataLineOutput implements AudioOutput { + private final SourceDataLine speaker; - Thread playbackThread = playbackThreadRef.getAndSet(null); - if (playbackThread != null) { - playbackThread.interrupt(); - try { - playbackThread.join(500); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + private SourceDataLineOutput(SourceDataLine speaker) { + this.speaker = speaker; + } + + @Override + public boolean isOpen() { + return speaker.isOpen(); + } + + @Override + public int write(byte[] audioData, int offset, int length) { + return speaker.write(audioData, offset, length); + } + + @Override + public void drain() { + speaker.drain(); } - SourceDataLine speaker = speakerRef.getAndSet(null); - if (speaker != null) { + @Override + public void stop() { speaker.stop(); + } + + @Override + public void flush() { + speaker.flush(); + } + + @Override + public void close() { speaker.close(); } - System.out.println("šŸ”Š Audio playback stopped"); } + enum PlaybackState { + RUNNING, + DRAIN_REQUESTED, + COMPLETED, + ABORTED + } + + private static final byte[] DRAIN_MARKER = new byte[0]; + private static final byte[] ABORT_MARKER = new byte[0]; + /** - * Handle incoming server events. Queues audio chunks for playback and signals completion - * when the response is finished or an error is reported. - * - * @param event The server event - * @param audioQueue Queue to receive audio data - * @param completionLatch Latch to release when the response is complete or fails + * Owns the playback worker and signals completion only after every queued packet is written and drained. */ - private static void handleEvent(SessionServerEvent event, BlockingQueue audioQueue, - CountDownLatch completionLatch) { - ServerEventType eventType = event.getType(); + static final class PlaybackController implements AutoCloseable { + private final AudioOutput output; + private final Object lifecycleLock = new Object(); + private final BlockingQueue queue; + private final Semaphore audioQueueSlots; + private final CompletableFuture completion = new CompletableFuture<>(); + private final AtomicReference state = new AtomicReference<>(PlaybackState.RUNNING); + private final AtomicLong acceptedAudioBytes = new AtomicLong(); + private final AtomicLong droppedAudioChunks = new AtomicLong(); + private final Thread playbackThread; + + PlaybackController(AudioOutput output, int audioQueueCapacity) { + if (audioQueueCapacity <= 0) { + throw new IllegalArgumentException("audioQueueCapacity must be positive"); + } + this.output = output; + // The semaphore bounds audio packets while the command queue always accepts terminal markers. + this.queue = new LinkedBlockingQueue<>(); + this.audioQueueSlots = new Semaphore(audioQueueCapacity); + this.playbackThread = new Thread(this::playbackLoop, "AudioPlayback"); + this.playbackThread.setDaemon(true); + } - if (eventType == ServerEventType.SESSION_CREATED) { - System.out.println("āœ“ Session created"); - } else if (eventType == ServerEventType.SESSION_UPDATED) { - System.out.println("āœ“ Session updated - ready to receive audio"); - } else if (eventType == ServerEventType.RESPONSE_AUDIO_DELTA) { - // Receive audio response and queue for playback - if (event instanceof SessionUpdateResponseAudioDelta) { - SessionUpdateResponseAudioDelta audioEvent = (SessionUpdateResponseAudioDelta) event; - byte[] audioData = audioEvent.getDelta(); - if (audioData != null && audioData.length > 0) { - if (!audioQueue.offer(audioData)) { - System.err.println("Warning: audio queue full, dropping chunk of " + audioData.length + " bytes"); - } else { - System.out.println("šŸ”Š Received audio chunk: " + audioData.length + " bytes"); + void start() { + playbackThread.start(); + } + + boolean queueAudio(byte[] audioData) { + if (audioData == null || audioData.length == 0) { + return false; + } + synchronized (lifecycleLock) { + if (state.get() != PlaybackState.RUNNING) { + return false; + } + if (!audioQueueSlots.tryAcquire()) { + droppedAudioChunks.incrementAndGet(); + return false; + } + + queue.offer(audioData); + acceptedAudioBytes.addAndGet(audioData.length); + return true; + } + } + + Mono finishGracefully() { + synchronized (lifecycleLock) { + if (state.compareAndSet(PlaybackState.RUNNING, PlaybackState.DRAIN_REQUESTED)) { + queue.offer(DRAIN_MARKER); + } + if (state.get() == PlaybackState.ABORTED) { + return Mono.error(new IllegalStateException("Playback was aborted")); + } + } + return Mono.fromFuture(completion); + } + + void abort(Throwable cause) { + synchronized (lifecycleLock) { + PlaybackState current = state.get(); + if (current == PlaybackState.COMPLETED || current == PlaybackState.ABORTED) { + return; + } + state.set(PlaybackState.ABORTED); + queue.clear(); + queue.offer(ABORT_MARKER); + } + + shutdownOutput(true); + playbackThread.interrupt(); + completion.completeExceptionally(cause == null + ? new IllegalStateException("Playback aborted") + : cause); + } + + long getAcceptedAudioBytes() { + return acceptedAudioBytes.get(); + } + + long getDroppedAudioChunks() { + return droppedAudioChunks.get(); + } + + PlaybackState getState() { + return state.get(); + } + + Mono completion() { + return Mono.fromFuture(completion); + } + + private void playbackLoop() { + Throwable failure = null; + try { + while (true) { + byte[] audioData = queue.take(); + if (audioData == DRAIN_MARKER) { + output.drain(); + synchronized (lifecycleLock) { + state.compareAndSet(PlaybackState.DRAIN_REQUESTED, PlaybackState.COMPLETED); + } + break; + } + if (audioData == ABORT_MARKER) { + break; } + + audioQueueSlots.release(); + if (state.get() != PlaybackState.ABORTED) { + writeFully(audioData); + } + } + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + if (state.get() != PlaybackState.ABORTED) { + failure = error; + state.set(PlaybackState.ABORTED); } + } catch (Throwable error) { + failure = error; + state.set(PlaybackState.ABORTED); + } finally { + shutdownOutput(state.get() == PlaybackState.ABORTED); + if (state.get() == PlaybackState.COMPLETED) { + completion.complete(null); + } else if (!completion.isDone()) { + completion.completeExceptionally(failure == null + ? new IllegalStateException("Playback aborted before drain completed") + : failure); + } + } + } + + private void writeFully(byte[] audioData) { + if (!output.isOpen()) { + throw new IllegalStateException("Speaker closed before queued audio was written"); + } + int bytesWritten = output.write(audioData, 0, audioData.length); + if (bytesWritten != audioData.length) { + throw new IllegalStateException( + "Speaker wrote " + bytesWritten + " of " + audioData.length + " queued bytes"); + } + } + + private void shutdownOutput(boolean flush) { + if (flush) { + try { + output.flush(); + } catch (Exception ignored) { + // Best-effort abort cleanup. + } + } + try { + output.stop(); + } catch (Exception ignored) { + // Best-effort cleanup. + } + try { + output.close(); + } catch (Exception ignored) { + // Best-effort cleanup. + } + } + + @Override + public void close() { + PlaybackState currentState = state.get(); + if (currentState != PlaybackState.COMPLETED && currentState != PlaybackState.ABORTED) { + abort(new IllegalStateException("Playback closed before graceful completion")); } - } else if (eventType == ServerEventType.RESPONSE_AUDIO_DONE) { - System.out.println("āœ“ Audio response complete"); - } else if (eventType == ServerEventType.RESPONSE_DONE) { - System.out.println("āœ“ Response complete"); - completionLatch.countDown(); - } else if (eventType == ServerEventType.ERROR) { - System.err.println("āŒ Error occurred in session " - + ((SessionUpdateError) event).getError().getMessage()); - completionLatch.countDown(); } } - // Private constructor to prevent instantiation private AudioPlaybackSample() { } } diff --git a/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/VoiceAssistantSample.java b/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/VoiceAssistantSample.java index 315df9ca0668..eb11273951b9 100644 --- a/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/VoiceAssistantSample.java +++ b/sdk/voicelive/azure-ai-voicelive/src/samples/java/com/azure/ai/voicelive/VoiceAssistantSample.java @@ -15,9 +15,13 @@ import com.azure.ai.voicelive.models.OutputAudioFormat; import com.azure.ai.voicelive.models.ServerEventType; import com.azure.ai.voicelive.models.ServerVadTurnDetection; +import com.azure.ai.voicelive.models.SessionResponse; +import com.azure.ai.voicelive.models.SessionResponseStatus; import com.azure.ai.voicelive.models.SessionServerEvent; import com.azure.ai.voicelive.models.SessionUpdateError; import com.azure.ai.voicelive.models.SessionUpdateResponseAudioDelta; +import com.azure.ai.voicelive.models.SessionUpdateResponseAudioDone; +import com.azure.ai.voicelive.models.SessionUpdateResponseDone; import com.azure.ai.voicelive.models.SessionUpdateSessionUpdated; import com.azure.ai.voicelive.models.VoiceLiveSessionOptions; import com.azure.core.credential.TokenCredential; @@ -33,12 +37,16 @@ import javax.sound.sampled.SourceDataLine; import javax.sound.sampled.TargetDataLine; +import java.time.Duration; import java.util.Arrays; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; /** @@ -134,22 +142,30 @@ private static class AudioPlaybackPacket { * *

Supports interruption handling where user speech can cancel ongoing assistant responses.

*/ - private static class AudioProcessor { + static final class AudioProcessor { private final VoiceLiveSessionAsyncClient session; private final AudioFormat audioFormat; // Audio capture components // volatile: shared between the reactor event thread (startCapture) and the audio capture worker thread private volatile TargetDataLine microphone; + private volatile Thread captureThread; private final AtomicBoolean isCapturing = new AtomicBoolean(false); // Audio playback components // volatile: shared between the reactor event thread (startPlayback) and the audio playback worker thread private volatile SourceDataLine speaker; + private volatile Thread playbackThread; + private final CountDownLatch playbackCompleted = new CountDownLatch(1); + private final AtomicReference playbackFailure = new AtomicReference<>(); + private final CompletableFuture playbackTermination = new CompletableFuture<>(); + private final Object playbackControlLock = new Object(); private final BlockingQueue playbackQueue = new LinkedBlockingQueue<>(1000); private final AtomicBoolean isPlaying = new AtomicBoolean(false); private final AtomicInteger nextSequenceNumber = new AtomicInteger(0); private final AtomicInteger playbackBase = new AtomicInteger(0); + private final VoiceAssistantPlaybackDiagnostics playbackDiagnostics + = new VoiceAssistantPlaybackDiagnostics(); AudioProcessor(VoiceLiveSessionAsyncClient session) { this.session = session; @@ -186,7 +202,7 @@ void startCapture() { isCapturing.set(true); // Start capture thread - Thread captureThread = new Thread(this::captureAudioLoop, "VoiceLive-AudioCapture"); + captureThread = new Thread(this::captureAudioLoop, "VoiceLive-AudioCapture"); captureThread.setDaemon(true); captureThread.start(); @@ -220,7 +236,7 @@ void startPlayback() { isPlaying.set(true); // Start playback thread - Thread playbackThread = new Thread(this::playbackAudioLoop, "VoiceLive-AudioPlayback"); + playbackThread = new Thread(this::playbackAudioLoop, "VoiceLive-AudioPlayback"); playbackThread.setDaemon(true); playbackThread.start(); @@ -271,82 +287,193 @@ private void captureAudioLoop() { * Audio playback loop - runs in separate thread */ private void playbackAudioLoop() { - while (isPlaying.get()) { - try { + try { + while (true) { AudioPlaybackPacket packet = playbackQueue.take(); // Blocking wait if (packet.audioData == null) { - // Shutdown signal + synchronized (playbackControlLock) { + if (isPlaying.get() && speaker != null && speaker.isOpen()) { + speaker.drain(); + } + } break; } - // Check if packet should be skipped (interrupted) - int currentBase = playbackBase.get(); - if (packet.sequenceNumber < currentBase) { - // Skip interrupted audio - continue; - } + synchronized (playbackControlLock) { + // Check and write under the same lock so interruption cannot flush between them. + int currentBase = playbackBase.get(); + if (packet.sequenceNumber < currentBase) { + playbackDiagnostics.recordSkippedPacket(); + continue; + } - // Play the audio - if (speaker != null && speaker.isOpen()) { - speaker.write(packet.audioData, 0, packet.audioData.length); + if (speaker != null && speaker.isOpen()) { + speaker.write(packet.audioData, 0, packet.audioData.length); + } } - - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } catch (Exception e) { - System.err.println("āŒ Error in audio playback: " + e.getMessage()); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (isPlaying.get()) { + playbackFailure.compareAndSet(null, e); + } + } catch (Exception e) { + playbackFailure.compareAndSet(null, e); + System.err.println("āŒ Error in audio playback: " + e.getMessage()); + } finally { + Throwable failure = playbackFailure.get(); + if (failure == null) { + playbackTermination.complete(null); + } else { + playbackTermination.completeExceptionally(failure); + } + playbackCompleted.countDown(); } } + Mono playbackFailure() { + return Mono.fromFuture(playbackTermination).then(Mono.never()); + } + /** * Queue audio data for playback */ void queueAudio(byte[] audioData) { - if (audioData != null && audioData.length > 0) { - int seqNum = nextSequenceNumber.getAndIncrement(); - // offer() returns false if the bounded queue is full; warn so a slow consumer is visible - if (!playbackQueue.offer(new AudioPlaybackPacket(seqNum, audioData))) { - System.err.println("Warning: playback queue full, dropping audio packet seq=" + seqNum); - } + if (audioData == null || audioData.length == 0) { + return; } + + int seqNum = nextSequenceNumber.getAndIncrement(); + // offer() returns false if the bounded queue is full; count drops without logging every delta. + boolean accepted = playbackQueue.offer(new AudioPlaybackPacket(seqNum, audioData)); + playbackDiagnostics.recordAudioChunk(audioData.length, accepted, playbackQueue.size()); } /** * Skip pending audio (for interruption handling) */ void skipPendingAudio() { - playbackBase.set(nextSequenceNumber.get()); - playbackQueue.clear(); + int cutoff; + int removed = 0; + int queueDepth; + synchronized (playbackControlLock) { + cutoff = nextSequenceNumber.get(); + playbackBase.set(cutoff); + + AudioPlaybackPacket packet; + while ((packet = playbackQueue.poll()) != null) { + if (packet.audioData != null) { + removed++; + } + } + queueDepth = playbackQueue.size(); + + // Flush after advancing the cutoff and clearing queued audio. + if (speaker != null && speaker.isOpen()) { + speaker.flush(); + } + } + playbackDiagnostics.recordSkip(removed); + + System.out.println("Playback interruption: skipCount=" + playbackDiagnostics.getSkipOperationCount() + + ", removed=" + removed + + ", cutoff=" + cutoff + + ", totalSkipped=" + playbackDiagnostics.getSkippedPacketCount() + + ", queueCurrent=" + queueDepth + + ", queueHighWaterApprox=" + playbackDiagnostics.getHighWaterQueueDepth()); + } + + void printAudioSummary(SessionUpdateResponseAudioDone audioDone) { + System.out.println("Audio response complete: responseId=" + audioDone.getResponseId() + + ", itemId=" + audioDone.getItemId() + + ", outputIndex=" + audioDone.getOutputIndex() + + ", contentIndex=" + audioDone.getContentIndex() + + ", chunks=" + playbackDiagnostics.getAudioChunkCount() + + ", bytes=" + playbackDiagnostics.getAudioByteCount() + + ", dropped=" + playbackDiagnostics.getDroppedPacketCount() + + ", skipped=" + playbackDiagnostics.getSkippedPacketCount() + + ", queueCurrent=" + playbackQueue.size() + + ", queueHighWaterApprox=" + playbackDiagnostics.getHighWaterQueueDepth()); + } - // Also drain the speaker buffer to stop playback immediately - if (speaker != null && speaker.isOpen()) { - speaker.flush(); + void printResponseSummary(String responseId, SessionResponseStatus status) { + System.out.println("āœ… Response complete: responseId=" + responseId + + ", status=" + status + + ", chunks=" + playbackDiagnostics.getAudioChunkCount() + + ", bytes=" + playbackDiagnostics.getAudioByteCount() + + ", dropped=" + playbackDiagnostics.getDroppedPacketCount() + + ", skipped=" + playbackDiagnostics.getSkippedPacketCount() + + ", skipCalls=" + playbackDiagnostics.getSkipOperationCount() + + ", queueCurrent=" + playbackQueue.size() + + ", queueHighWaterApprox=" + playbackDiagnostics.getHighWaterQueueDepth()); + } + + /** + * Stop capture and drain queued playback after a normal receive-stream completion. + */ + void shutdownGracefully() { + stopCapture(); + if (isPlaying.get() && playbackThread != null) { + try { + if (!playbackQueue.offer(new AudioPlaybackPacket(-1, null), 5, TimeUnit.SECONDS)) { + throw new IllegalStateException("Timed out enqueueing playback drain marker"); + } + if (!playbackCompleted.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("Timed out waiting for queued audio to drain"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while draining queued audio", e); + } + Throwable failure = playbackFailure.get(); + if (failure != null) { + throw new IllegalStateException("Audio playback failed", failure); + } } + isPlaying.set(false); + closeSpeaker(false); } /** - * Stop capture and playback + * Abort capture and playback without waiting for queued audio. */ void shutdown() { - // Stop capture + stopCapture(); + isPlaying.set(false); + playbackQueue.clear(); + playbackQueue.offer(new AudioPlaybackPacket(-1, null)); + Thread currentPlaybackThread = playbackThread; + if (currentPlaybackThread != null) { + currentPlaybackThread.interrupt(); + } + closeSpeaker(true); + } + + private void stopCapture() { isCapturing.set(false); if (microphone != null) { microphone.stop(); microphone.close(); microphone = null; } + Thread currentCaptureThread = captureThread; + if (currentCaptureThread != null) { + currentCaptureThread.interrupt(); + } System.out.println("šŸŽ¤ Microphone capture stopped"); + } - // Stop playback - isPlaying.set(false); - playbackQueue.offer(new AudioPlaybackPacket(-1, null)); // Shutdown signal - if (speaker != null) { - speaker.stop(); - speaker.close(); - speaker = null; + private void closeSpeaker(boolean flush) { + synchronized (playbackControlLock) { + if (speaker != null) { + if (flush && speaker.isOpen()) { + speaker.flush(); + } + speaker.stop(); + speaker.close(); + speaker = null; + } } System.out.println("šŸ”Š Audio playback stopped"); } @@ -455,39 +582,44 @@ private static void runVoiceAssistantWithClient(VoiceLiveAsyncClient client) { System.out.println("āœ“ VoiceLive client created"); AtomicReference audioProcessorRef = new AtomicReference<>(); - - // Latch keeps main alive until the event stream completes (or an error occurs). - final CountDownLatch completionLatch = new CountDownLatch(1); - - // Start session. Session lifetime is local to this reactive chain ā€” the session is - // captured by the lambda passed to flatMapMany and then threaded into per-event handling - // via flatMap, so no instance field or shared holder is needed. - client.startSession(DEFAULT_MODEL, null) - .flatMapMany(session -> { - System.out.println("āœ“ Session started successfully"); - audioProcessorRef.set(new AudioProcessor(session)); - return configureSession(session) - .thenMany(session.receiveEvents()) - .flatMap(event -> handleServerEvent(event, audioProcessorRef.get())); - }) - .subscribe( - ignored -> { }, - error -> { - System.err.println("āŒ Error receiving events: " + error.getMessage()); - shutdownAudio(audioProcessorRef); - completionLatch.countDown(); - }, - () -> { - System.out.println("āœ“ Event stream completed"); - shutdownAudio(audioProcessorRef); - completionLatch.countDown(); - } - ); - + AtomicReference sessionRef = new AtomicReference<>(); + Thread shutdownHook = new Thread(() -> { + shutdownAudio(audioProcessorRef); + closeSession(sessionRef); + }, "VoiceLive-Shutdown"); + Runtime.getRuntime().addShutdownHook(shutdownHook); + + boolean streamCompleted = false; try { - completionLatch.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + Mono.usingWhen( + client.startSession(DEFAULT_MODEL, null).doOnNext(session -> { + System.out.println("āœ“ Session started successfully"); + sessionRef.set(session); + audioProcessorRef.set(new AudioProcessor(session)); + }), + session -> { + AudioProcessor audioProcessor = audioProcessorRef.get(); + Mono eventStream = configureSession(session) + .thenMany(session.receiveEvents()) + .doOnNext(event -> handleServerEvent(event, audioProcessor)) + .then(); + return Mono.firstWithSignal(eventStream, audioProcessor.playbackFailure()); + }, + session -> closeSessionAsync(session, sessionRef)) + .block(); + shutdownAudioGracefully(audioProcessorRef); + streamCompleted = true; + System.out.println("āœ“ Event stream completed"); + } finally { + if (!streamCompleted) { + shutdownAudio(audioProcessorRef); + } + closeSession(sessionRef); + try { + Runtime.getRuntime().removeShutdownHook(shutdownHook); + } catch (IllegalStateException ignored) { + // JVM shutdown is already in progress and the hook is running. + } } } @@ -500,7 +632,18 @@ private static Mono configureSession(VoiceLiveSessionAsyncClient session) } /** - * Cleanup audio processor. + * Stop capture and drain playback after a normal stream completion. + */ + private static void shutdownAudioGracefully(AtomicReference audioProcessorRef) { + AudioProcessor audioProcessor = audioProcessorRef.get(); + if (audioProcessor != null) { + audioProcessor.shutdownGracefully(); + audioProcessorRef.compareAndSet(audioProcessor, null); + } + } + + /** + * Abort the audio processor. */ private static void shutdownAudio(AtomicReference audioProcessorRef) { AudioProcessor audioProcessor = audioProcessorRef.getAndSet(null); @@ -509,6 +652,24 @@ private static void shutdownAudio(AtomicReference audioProcessor } } + private static Mono closeSessionAsync(VoiceLiveSessionAsyncClient session, + AtomicReference sessionRef) { + return session.closeAsync() + .timeout(Duration.ofSeconds(5)) + .doOnSuccess(ignored -> sessionRef.compareAndSet(session, null)); + } + + private static void closeSession(AtomicReference sessionRef) { + VoiceLiveSessionAsyncClient session = sessionRef.getAndSet(null); + if (session != null) { + try { + session.closeAsync().block(Duration.ofSeconds(5)); + } catch (Exception error) { + System.err.println("āŒ Error closing session: " + error.getMessage()); + } + } + } + /** * Create session configuration for voice conversation */ @@ -546,56 +707,113 @@ private static VoiceLiveSessionOptions createVoiceSessionOptions() { } /** - * Handle a single server event. Returns a {@link Mono} so the per-event handling stays - * inside the reactive chain (no nested subscribe). The voice assistant doesn't send any - * follow-up events, so handlers always return {@link Mono#empty()}. + * Handle a single server event. Exceptions propagate through the receive stream. */ - private static Mono handleServerEvent(SessionServerEvent event, AudioProcessor audioProcessor) { + static void handleServerEvent(SessionServerEvent event, AudioProcessor audioProcessor) { ServerEventType eventType = event.getType(); - try { - if (eventType == ServerEventType.SESSION_CREATED) { - System.out.println("āœ“ Session created - initializing..."); - } else if (event instanceof SessionUpdateSessionUpdated) { - System.out.println("āœ“ Session updated - starting audio"); - - // Print the full JSON representation - SessionUpdateSessionUpdated sessionUpdated = (SessionUpdateSessionUpdated) event; - System.out.println("šŸ“„ Session Updated Event (Full JSON):"); - System.out.println(BinaryData.fromObject(sessionUpdated).toString()); - - audioProcessor.startPlayback(); - audioProcessor.startCapture(); - - System.out.println("šŸŽ¤ VOICE ASSISTANT READY"); - System.out.println("Start speaking to begin conversation"); - System.out.println("Press Ctrl+C to exit"); - } else if (eventType == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STARTED) { - System.out.println("šŸŽ¤ Speech detected"); - // Server handles interruption automatically with interruptResponse=true - // Just clear any pending audio in the playback queue - audioProcessor.skipPendingAudio(); - } else if (eventType == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STOPPED) { - System.out.println("šŸ¤” Speech ended - processing..."); - } else if (event instanceof SessionUpdateResponseAudioDelta) { - SessionUpdateResponseAudioDelta audioEvent = (SessionUpdateResponseAudioDelta) event; - byte[] audioData = audioEvent.getDelta(); - if (audioData != null && audioData.length > 0) { - audioProcessor.queueAudio(audioData); - } - } else if (eventType == ServerEventType.RESPONSE_AUDIO_DONE) { - System.out.println("šŸŽ¤ Ready for next input..."); - } else if (eventType == ServerEventType.RESPONSE_DONE) { - System.out.println("āœ… Response complete"); - } else if (event instanceof SessionUpdateError) { - SessionUpdateError errorEvent = (SessionUpdateError) event; - System.out.println("āŒ VoiceLive error: " + errorEvent.getError().getMessage()); + if (eventType == ServerEventType.SESSION_CREATED) { + System.out.println("āœ“ Session created - initializing..."); + } else if (event instanceof SessionUpdateSessionUpdated) { + System.out.println("āœ“ Session updated - starting audio"); + + // Print the full JSON representation + SessionUpdateSessionUpdated sessionUpdated = (SessionUpdateSessionUpdated) event; + System.out.println("šŸ“„ Session Updated Event (Full JSON):"); + System.out.println(BinaryData.fromObject(sessionUpdated).toString()); + + audioProcessor.startPlayback(); + audioProcessor.startCapture(); + + System.out.println("šŸŽ¤ VOICE ASSISTANT READY"); + System.out.println("Start speaking to begin conversation"); + System.out.println("Press Ctrl+C to exit"); + } else if (eventType == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STARTED) { + System.out.println("šŸŽ¤ Speech detected"); + // Server handles interruption automatically with interruptResponse=true. + // Preserve immediate queue clearing so pending assistant audio is not played. + audioProcessor.skipPendingAudio(); + } else if (eventType == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STOPPED) { + System.out.println("šŸ¤” Speech ended - processing..."); + } else if (event instanceof SessionUpdateResponseAudioDelta) { + audioProcessor.queueAudio(((SessionUpdateResponseAudioDelta) event).getDelta()); + } else if (event instanceof SessionUpdateResponseAudioDone) { + audioProcessor.printAudioSummary((SessionUpdateResponseAudioDone) event); + System.out.println("šŸŽ¤ Ready for next input..."); + } else if (event instanceof SessionUpdateResponseDone) { + SessionResponse response = ((SessionUpdateResponseDone) event).getResponse(); + String responseId = response == null ? null : response.getId(); + SessionResponseStatus status = response == null ? null : response.getStatus(); + if (audioProcessor == null) { + System.out.println("āœ… Response complete: responseId=" + responseId + ", status=" + status); + } else { + audioProcessor.printResponseSummary(responseId, status); } - } catch (Exception e) { - System.err.println("āŒ Error handling event: " + e.getMessage()); - e.printStackTrace(); + } else if (event instanceof SessionUpdateError) { + SessionUpdateError errorEvent = (SessionUpdateError) event; + String message = errorEvent.getError() == null + ? "Unknown VoiceLive error" + : errorEvent.getError().getMessage(); + throw new IllegalStateException("VoiceLive error: " + message); + } + } +} + +/** + * Hardware-independent aggregate playback diagnostics used by {@link VoiceAssistantSample}. + */ +final class VoiceAssistantPlaybackDiagnostics { + private final AtomicLong audioChunkCount = new AtomicLong(0); + private final AtomicLong audioByteCount = new AtomicLong(0); + private final AtomicLong droppedPacketCount = new AtomicLong(0); + private final AtomicLong skippedPacketCount = new AtomicLong(0); + private final AtomicLong skipOperationCount = new AtomicLong(0); + private final AtomicInteger highWaterQueueDepth = new AtomicInteger(0); + + void recordAudioChunk(int byteCount, boolean accepted, int queueDepth) { + audioChunkCount.incrementAndGet(); + audioByteCount.addAndGet(byteCount); + if (!accepted) { + droppedPacketCount.incrementAndGet(); } + recordQueueDepth(queueDepth); + } + + void recordQueueDepth(int queueDepth) { + int normalizedQueueDepth = Math.max(0, queueDepth); + highWaterQueueDepth.updateAndGet(current -> Math.max(current, normalizedQueueDepth)); + } + + void recordSkippedPacket() { + skippedPacketCount.incrementAndGet(); + } + + void recordSkip(int removed) { + skippedPacketCount.addAndGet(removed); + skipOperationCount.incrementAndGet(); + } + + long getAudioChunkCount() { + return audioChunkCount.get(); + } + + long getAudioByteCount() { + return audioByteCount.get(); + } + + long getDroppedPacketCount() { + return droppedPacketCount.get(); + } + + long getSkippedPacketCount() { + return skippedPacketCount.get(); + } + + long getSkipOperationCount() { + return skipOperationCount.get(); + } - return Mono.empty(); + int getHighWaterQueueDepth() { + return highWaterQueueDepth.get(); } } diff --git a/sdk/voicelive/azure-ai-voicelive/src/test/java/com/azure/ai/voicelive/AudioSampleLifecycleTest.java b/sdk/voicelive/azure-ai-voicelive/src/test/java/com/azure/ai/voicelive/AudioSampleLifecycleTest.java new file mode 100644 index 000000000000..5ffc345aca87 --- /dev/null +++ b/sdk/voicelive/azure-ai-voicelive/src/test/java/com/azure/ai/voicelive/AudioSampleLifecycleTest.java @@ -0,0 +1,208 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.ai.voicelive; + +import com.azure.ai.voicelive.models.SessionResponseStatus; +import com.azure.ai.voicelive.models.SessionServerEvent; +import com.azure.core.util.BinaryData; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class AudioSampleLifecycleTest { + + @Test + void playbackCompletesOnlyAfterQueuedAudioIsWrittenAndDrained() { + TestAudioOutput output = new TestAudioOutput(false); + AudioPlaybackSample.PlaybackController playback = new AudioPlaybackSample.PlaybackController(output, 2); + playback.start(); + + assertTrue(playback.queueAudio(new byte[] { 1, 2 })); + assertTrue(playback.queueAudio(new byte[] { 3, 4 })); + playback.finishGracefully().block(Duration.ofSeconds(5)); + + assertArrayEquals(new byte[] { 1, 2, 3, 4 }, output.getWrittenAudio()); + assertEquals(4, playback.getAcceptedAudioBytes()); + assertEquals(AudioPlaybackSample.PlaybackState.COMPLETED, playback.getState()); + assertTrue(output.indexOf("drain") > output.lastIndexOf("write")); + assertTrue(output.indexOf("close") > output.indexOf("drain")); + } + + @Test + void terminalMarkerRemainsReliableWhenAudioQueueIsFull() throws InterruptedException { + TestAudioOutput output = new TestAudioOutput(true); + AudioPlaybackSample.PlaybackController playback = new AudioPlaybackSample.PlaybackController(output, 1); + playback.start(); + + assertTrue(playback.queueAudio(new byte[] { 1 })); + assertTrue(output.awaitWriteStarted()); + assertTrue(playback.queueAudio(new byte[] { 2 })); + assertFalse(playback.queueAudio(new byte[] { 3 })); + assertEquals(1, playback.getDroppedAudioChunks()); + + // The one audio slot is full, but the terminal marker must still be accepted. + playback.finishGracefully(); + output.releaseWrite(); + playback.completion().block(Duration.ofSeconds(5)); + + assertArrayEquals(new byte[] { 1, 2 }, output.getWrittenAudio()); + assertEquals(AudioPlaybackSample.PlaybackState.COMPLETED, playback.getState()); + assertEquals(1, output.count("drain")); + } + + @Test + void abortDiscardsQueuedAudioAndDoesNotDrain() throws InterruptedException { + TestAudioOutput output = new TestAudioOutput(true); + AudioPlaybackSample.PlaybackController playback = new AudioPlaybackSample.PlaybackController(output, 2); + playback.start(); + + assertTrue(playback.queueAudio(new byte[] { 1 })); + assertTrue(output.awaitWriteStarted()); + assertTrue(playback.queueAudio(new byte[] { 2 })); + + playback.abort(new IllegalStateException("test abort")); + + assertThrows(RuntimeException.class, () -> playback.completion().block(Duration.ofSeconds(5))); + assertEquals(AudioPlaybackSample.PlaybackState.ABORTED, playback.getState()); + assertFalse(output.getEvents().contains("drain")); + assertTrue(output.getEvents().contains("flush")); + assertTrue(output.getEvents().contains("close")); + } + + @Test + void completedResponseRequiresCompletedStatusAndAudio() { + AudioPlaybackSample.validateCompletedResponse("response-1", SessionResponseStatus.COMPLETED, 1); + + assertThrows(IllegalStateException.class, + () -> AudioPlaybackSample.validateCompletedResponse("response-2", SessionResponseStatus.CANCELLED, 1)); + assertThrows(IllegalStateException.class, + () -> AudioPlaybackSample.validateCompletedResponse("response-3", null, 1)); + assertThrows(IllegalStateException.class, + () -> AudioPlaybackSample.validateCompletedResponse("response-4", SessionResponseStatus.COMPLETED, 0)); + } + + @Test + void assistantHandlesResponseDoneWithMissingResponseSafely() { + SessionServerEvent responseDone + = BinaryData.fromString("{\"type\":\"response.done\"," + "\"event_id\":\"event-3\",\"response\":null}") + .toObject(SessionServerEvent.class); + + VoiceAssistantSample.handleServerEvent(responseDone, null); + } + + @Test + void assistantTracksAggregateQueueDropAndSkipDiagnosticsWithoutHardware() { + VoiceAssistantPlaybackDiagnostics diagnostics = new VoiceAssistantPlaybackDiagnostics(); + + for (int i = 0; i < 1000; i++) { + diagnostics.recordAudioChunk(2, true, i + 1); + } + diagnostics.recordAudioChunk(2, false, 1000); + + assertEquals(1001, diagnostics.getAudioChunkCount()); + assertEquals(2002, diagnostics.getAudioByteCount()); + assertEquals(1, diagnostics.getDroppedPacketCount()); + assertEquals(1000, diagnostics.getHighWaterQueueDepth()); + + diagnostics.recordSkip(1000); + + assertEquals(1, diagnostics.getSkipOperationCount()); + assertEquals(1000, diagnostics.getSkippedPacketCount()); + } + + private static final class TestAudioOutput implements AudioPlaybackSample.AudioOutput { + private volatile boolean open = true; + private final List events = new CopyOnWriteArrayList<>(); + private final ByteArrayOutputStream writtenAudio = new ByteArrayOutputStream(); + private final CountDownLatch writeStarted = new CountDownLatch(1); + private final CountDownLatch allowWrite; + + private TestAudioOutput(boolean blockWrites) { + this.allowWrite = new CountDownLatch(blockWrites ? 1 : 0); + } + + @Override + public boolean isOpen() { + return open; + } + + @Override + public int write(byte[] audioData, int offset, int length) { + events.add("write"); + writeStarted.countDown(); + try { + allowWrite.await(); + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + return 0; + } + synchronized (writtenAudio) { + writtenAudio.write(audioData, offset, length); + } + return length; + } + + @Override + public void drain() { + events.add("drain"); + } + + @Override + public void stop() { + events.add("stop"); + } + + @Override + public void flush() { + events.add("flush"); + } + + @Override + public void close() { + events.add("close"); + open = false; + } + + private boolean awaitWriteStarted() throws InterruptedException { + return writeStarted.await(5, TimeUnit.SECONDS); + } + + private void releaseWrite() { + allowWrite.countDown(); + } + + private byte[] getWrittenAudio() { + synchronized (writtenAudio) { + return writtenAudio.toByteArray(); + } + } + + private List getEvents() { + return events; + } + + private int indexOf(String event) { + return events.indexOf(event); + } + + private int lastIndexOf(String event) { + return events.lastIndexOf(event); + } + + private long count(String event) { + return events.stream().filter(event::equals).count(); + } + } +}