From 3c99843ebef9a5eaa764465baced5378c215d1f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Mon, 9 Nov 2026 16:02:45 +0800 Subject: [PATCH 1/2] feat(qwen-live): add Qwen Live realtime WebSocket SDK Add com.alibaba.dashscope.audio.live: QwenLiveConversation WebSocket session with single-connect lifecycle, connection/session config (QwenLiveConnectionParam, QwenLiveSessionConfig), client/responses delegation, byte[] audio input, unified onError with optional ordered async callback dispatch, and close/cancel/awaitSessionClosed semantics. Include protocol/conversation/config/param unit tests and quick-start and usage samples. QwenLiveConnectionParam reuses the existing DASHSCOPE_WEBSOCKET_OMNI_BASE_URL env override. --- samples/QwenLiveQuickStart.java | 162 ++++ samples/QwenLiveUsage.java | 311 +++++++ .../audio/live/QwenLiveCallback.java | 18 + .../live/QwenLiveCallbackDispatcher.java | 95 ++ .../audio/live/QwenLiveConnectionParam.java | 106 +++ .../audio/live/QwenLiveConstants.java | 71 ++ .../dashscope/audio/live/QwenLiveContext.java | 26 + .../audio/live/QwenLiveContextCompaction.java | 27 + .../audio/live/QwenLiveConversation.java | 477 ++++++++++ .../audio/live/QwenLiveDelegation.java | 35 + .../audio/live/QwenLiveDelegationType.java | 13 + .../audio/live/QwenLiveResponsesConfig.java | 31 + .../audio/live/QwenLiveSessionConfig.java | 86 ++ .../audio/live/QwenLiveTransport.java | 24 + .../TestQwenLiveConnectionParam.java | 182 ++++ .../dashscope/TestQwenLiveSessionConfig.java | 295 ++++++ .../audio/live/QwenLiveQuickStartChecks.java | 406 +++++++++ .../audio/live/TestQwenLiveConversation.java | 840 ++++++++++++++++++ .../audio/live/TestQwenLiveProtocol.java | 553 ++++++++++++ 19 files changed, 3758 insertions(+) create mode 100644 samples/QwenLiveQuickStart.java create mode 100644 samples/QwenLiveUsage.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallback.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallbackDispatcher.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConnectionParam.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConstants.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContext.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContextCompaction.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConversation.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegation.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegationType.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveResponsesConfig.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveSessionConfig.java create mode 100644 src/main/java/com/alibaba/dashscope/audio/live/QwenLiveTransport.java create mode 100644 src/test/java/com/alibaba/dashscope/TestQwenLiveConnectionParam.java create mode 100644 src/test/java/com/alibaba/dashscope/TestQwenLiveSessionConfig.java create mode 100644 src/test/java/com/alibaba/dashscope/audio/live/QwenLiveQuickStartChecks.java create mode 100644 src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveConversation.java create mode 100644 src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveProtocol.java diff --git a/samples/QwenLiveQuickStart.java b/samples/QwenLiveQuickStart.java new file mode 100644 index 0000000..836e380 --- /dev/null +++ b/samples/QwenLiveQuickStart.java @@ -0,0 +1,162 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.audio.live.*; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import java.io.BufferedOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.Base64; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +public class QwenLiveQuickStart { + private static final int INPUT_BYTES_PER_SECOND = 16000 * 2; + private static final int FRAME_BYTES = INPUT_BYTES_PER_SECOND / 10; + + public static void main(String[] args) throws Exception { + if (args.length < 4 || args.length > 6) { + System.out.println("Usage: QwenLiveQuickStart [idleSeconds=3] [timeoutSeconds=90]"); + System.exit(2); + return; + } + long idleMs = args.length > 4 ? Math.multiplyExact(Long.parseLong(args[4]), 1000) : 3000; + long timeoutMs = args.length > 5 ? Math.multiplyExact(Long.parseLong(args[5]), 1000) : 90000; + QwenLiveConnectionParam param = QwenLiveConnectionParam.builder().model("qwen-live").workspaceId(args[0]).build(); + try { + run(param, args[1], Paths.get(args[2]), Paths.get(args[3]), idleMs, timeoutMs); + System.out.println("Saved raw PCM: " + args[3] + " (stopped by audio idle window, not a server done event)"); + } catch (Exception failure) { + failure.printStackTrace(); + System.exit(1); + } + System.exit(0); + } + + static void run(QwenLiveConnectionParam param, String backendModel, Path input, Path output, + long idleMs, long timeoutMs) throws Exception { + if (idleMs <= 0 || timeoutMs <= idleMs) throw new IllegalArgumentException("Require 0 < idle < timeout."); + QwenLiveSessionConfig config = QwenLiveSessionConfig.builder() + .model("qwen-audio-realtime").instructions("用简洁中文回答用户的语音问题") + .inputAudioFormat("pcm").outputAudioFormat("pcm") + .context(QwenLiveContext.builder().compaction( + QwenLiveContextCompaction.builder().enabled(false).mode("local").build()).build()) + .delegation(QwenLiveDelegation.builder().type(QwenLiveDelegationType.RESPONSES) + .responses(QwenLiveResponsesConfig.builder().model(backendModel) + .instructions("直接回答用户问题，不使用外部工具").build()).build()).build(); + config.getConfig(); + CountDownLatch configured = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + AtomicLong lastAudio = new AtomicLong(Long.MIN_VALUE); + AtomicBoolean accepting = new AtomicBoolean(true); + try (OutputStream audioFile = new BufferedOutputStream( + Files.newOutputStream(output, StandardOpenOption.CREATE_NEW))) { + QwenLiveConversation conversation = new QwenLiveConversation(param, new QwenLiveCallback() { + @Override public void onEvent(JsonObject event) { + synchronized (audioFile) { + if (!accepting.get()) return; + try { + String type = event.has(FIELD_TYPE) ? event.get(FIELD_TYPE).getAsString() : ""; + if (EVENT_SESSION_UPDATED.equals(type)) configured.countDown(); + if (EVENT_SESSION_CLOSED.equals(type)) { + onError(new IOException("Session ended before audio idle window.")); + } + if (EVENT_SESSION_OUTPUT_AUDIO_DELTA.equals(type)) { + JsonElement audio = event.get(FIELD_AUDIO); + if (audio == null || audio.isJsonNull()) audio = event.get(FIELD_DELTA); + if (audio == null || audio.isJsonNull()) throw new IOException("Audio event has no audio/delta data."); + byte[] bytes = Base64.getDecoder().decode(audio.getAsString()); + if (bytes.length > 0) { + audioFile.write(bytes); + lastAudio.set(System.nanoTime()); + } + } + } catch (IOException | RuntimeException e) { onError(e); } + audioFile.notifyAll(); + } + } + + @Override public void onClose(int code, String reason) { + onError(new IOException("WebSocket closed before output settled: " + code)); + } + + @Override public void onError(Throwable cause) { + synchronized (audioFile) { + if (!accepting.get()) return; + failure.compareAndSet(null, cause); + configured.countDown(); + audioFile.notifyAll(); + } + } + }); + try { + conversation.connect(); + conversation.updateSession(config); + if (!configured.await(30, TimeUnit.SECONDS)) throw new TimeoutException("No session.updated confirmation."); + checkFailure(failure); + streamAudio(conversation, input, failure); + long inputFinished = System.nanoTime(); + synchronized (audioFile) { + while (true) { + checkFailure(failure); + long now = System.nanoTime(); + boolean receivedAudio = lastAudio.get() != Long.MIN_VALUE; + long idleLeft = TimeUnit.MILLISECONDS.toNanos(idleMs) + - (receivedAudio ? Math.min(now - lastAudio.get(), now - inputFinished) : 0); + long timeoutLeft = TimeUnit.MILLISECONDS.toNanos(timeoutMs) - (now - inputFinished); + if (receivedAudio && idleLeft <= 0) { accepting.set(false); break; } + if (timeoutLeft <= 0) throw new TimeoutException("Audio output did not settle; saved PCM may be incomplete."); + TimeUnit.NANOSECONDS.timedWait(audioFile, receivedAudio ? Math.min(idleLeft, timeoutLeft) : timeoutLeft); + } + } + conversation.close(); + } finally { + synchronized (audioFile) { accepting.set(false); } + conversation.cancel(); + } + } + } + + private static void streamAudio(QwenLiveConversation conversation, Path input, AtomicReference failure) + throws Exception { + byte[] frame = new byte[FRAME_BYTES]; + long started = System.nanoTime(); + long sentBytes = 0; + try (InputStream in = Files.newInputStream(input)) { + int read; + while ((read = in.read(frame)) != -1) { + checkFailure(failure); + conversation.appendAudioBytes(frame, 0, read); + sentBytes += read; + pace(started, sentBytes); + } + } + byte[] silence = new byte[FRAME_BYTES]; + for (int i = 0; i < 10; i++) { + checkFailure(failure); + conversation.appendAudioBytes(silence); + sentBytes += silence.length; + pace(started, sentBytes); + } + } + + private static void pace(long started, long sentBytes) throws InterruptedException { + long remaining = sentBytes * 1000000000L / INPUT_BYTES_PER_SECOND - (System.nanoTime() - started); + if (remaining > 0) TimeUnit.NANOSECONDS.sleep(remaining); + } + + private static void checkFailure(AtomicReference failure) { + Throwable cause = failure.get(); + if (cause != null) throw new IllegalStateException("Audio request failed; saved PCM may be incomplete.", cause); + } +} diff --git a/samples/QwenLiveUsage.java b/samples/QwenLiveUsage.java new file mode 100644 index 0000000..c489aa6 --- /dev/null +++ b/samples/QwenLiveUsage.java @@ -0,0 +1,311 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.audio.live.QwenLiveCallback; +import com.alibaba.dashscope.audio.live.QwenLiveSessionConfig; +import com.alibaba.dashscope.audio.live.QwenLiveContext; +import com.alibaba.dashscope.audio.live.QwenLiveContextCompaction; +import com.alibaba.dashscope.audio.live.QwenLiveConversation; +import com.alibaba.dashscope.audio.live.QwenLiveDelegation; +import com.alibaba.dashscope.audio.live.QwenLiveDelegationType; +import com.alibaba.dashscope.audio.live.QwenLiveConnectionParam; +import com.alibaba.dashscope.audio.live.QwenLiveResponsesConfig; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Base64; +import java.util.Collections; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +public class QwenLiveUsage { + private static final int FRAME_BYTES = 3200; + + private static final long FRAME_INTERVAL_MS = 100; + private static final int AWAIT_SECONDS = 30; + + private static final BlockingQueue delegations = new ArrayBlockingQueue<>(16); + private static final CountDownLatch sessionSettled = new CountDownLatch(1); + private static final AtomicBoolean sessionUpdated = new AtomicBoolean(false); + private static final AtomicBoolean running = new AtomicBoolean(true); + private static final AtomicLong outputAudioBytes = new AtomicLong(); + private static final AtomicReference usageSeconds = new AtomicReference<>(); + private static final StringBuilder inputTranscript = new StringBuilder(); + private static final StringBuilder outputTranscript = new StringBuilder(); + private static volatile QwenLiveConversation conversation; + + public static void main(String[] args) throws Exception { + if (args.length < 3 || args.length > 4) { + System.out.println( + "Usage: QwenLiveUsage [backendModel]"); + return; + } + String workspaceId = args[0]; + String mode = args[1]; + Path pcm = Paths.get(args[2]); + String backendModel = args.length == 4 ? args[3] : null; + if (!"client".equals(mode) && !"responses".equals(mode)) { + System.out.println("mode must be client or responses, got: " + mode); + return; + } + if ("responses".equals(mode) && (backendModel == null || backendModel.trim().isEmpty())) { + System.out.println("responses mode requires an explicit backendModel argument."); + return; + } + if (!Files.isRegularFile(pcm) || Files.size(pcm) == 0) { + System.out.println("pcmFile must be a non-empty regular file: " + pcm); + return; + } + if (hasWavHeader(pcm)) { + System.out.println( + "pcmFile looks like a WAV container; strip the header and pass raw PCM16."); + return; + } + + QwenLiveDelegation delegation; + if ("client".equals(mode)) { + delegation = QwenLiveDelegation.builder().type(QwenLiveDelegationType.CLIENT).build(); + } else { + delegation = + QwenLiveDelegation.builder() + .type(QwenLiveDelegationType.RESPONSES) + .responses( + QwenLiveResponsesConfig.builder() + .model(backendModel) + .instructions("根据任务调用已配置工具并返回清晰结果") + .tools(Collections.emptyList()) + .build()) + .build(); + } + QwenLiveSessionConfig config = + QwenLiveSessionConfig.builder() + .model("qwen-audio-realtime") + .instructions("用简洁中文与用户交流") + .inputAudioFormat("pcm") + .outputAudioFormat("pcm") + .context( + QwenLiveContext.builder() + .compaction( + QwenLiveContextCompaction.builder().enabled(false).mode("local").build()) + .build()) + .delegation(delegation) + .build(); + + ExecutorService agentWorker = Executors.newSingleThreadExecutor(); + ExecutorService callbackExecutor = Executors.newSingleThreadExecutor(); + try { + QwenLiveConnectionParam param = + QwenLiveConnectionParam.builder() + .model("qwen-live") + .workspaceId(workspaceId) + .callbackExecutor(callbackExecutor) + .callbackQueueCapacity(256) + .build(); + conversation = new QwenLiveConversation(param, new SampleCallback()); + conversation.connect(); + System.out.println("connected to " + param.getUrl()); + conversation.updateSession(config); + if (!sessionSettled.await(AWAIT_SECONDS, TimeUnit.SECONDS) || !sessionUpdated.get()) { + System.out.println("session.updated not confirmed; stop before sending audio."); + return; + } + if ("client".equals(mode)) { + agentWorker.submit(QwenLiveUsage::simulateExternalAgent); + } else { + conversation.createResponseItem(userTextItem("帮我看看今天适合户外活动吗")); + conversation.createResponse(); + } + streamAudio(pcm); + running.set(false); + agentWorker.shutdown(); + if (!agentWorker.awaitTermination(AWAIT_SECONDS, TimeUnit.SECONDS)) { + System.out.println("agent worker did not stop in time."); + } + conversation.close(); + System.out.println( + "session.closed received: " + conversation.awaitSessionClosed(AWAIT_SECONDS)); + System.out.println("session.closed reason: " + conversation.getCloseReason()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + System.out.println("interrupted: " + e.getMessage()); + } finally { + running.set(false); + if (conversation != null) conversation.cancel(); + agentWorker.shutdownNow(); + callbackExecutor.shutdown(); + boolean callbacksDrained = false; + try { + agentWorker.awaitTermination(AWAIT_SECONDS, TimeUnit.SECONDS); + callbacksDrained = callbackExecutor.awaitTermination(AWAIT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + if (!callbacksDrained) callbackExecutor.shutdownNow(); + } + System.out.println("session id: " + (conversation == null ? null : conversation.getSessionId())); + if (callbacksDrained) { + System.out.println("output audio bytes: " + outputAudioBytes.get()); + System.out.println("usage seconds (cumulative): " + usageSeconds.get()); + System.out.println("input transcript: " + inputTranscript); + System.out.println("output transcript: " + outputTranscript); + } else { + System.out.println("callbacks did not finish; final statistics are unavailable."); + } + } + } + + private static void streamAudio(Path pcm) throws Exception { + byte[] frame = new byte[FRAME_BYTES]; + long frames = 0; + try (InputStream in = Files.newInputStream(pcm)) { + int read; + while (running.get() && (read = in.read(frame)) > 0) { + conversation.appendAudioBytes(frame, 0, read); + frames++; + Thread.sleep(FRAME_INTERVAL_MS); + } + } + System.out.println("sent audio frames: " + frames); + } + + private static void simulateExternalAgent() { + try { + while (running.get()) { + JsonObject event = delegations.poll(1, TimeUnit.SECONDS); + if (event == null) { + continue; + } + String delegationId = text(event, FIELD_DELEGATION_ID); + if (delegationId == null) { + System.out.println("delegation event without delegation_id, skipped."); + continue; + } + System.out.println("delegation arguments: " + text(event, FIELD_ARGUMENTS)); + conversation.appendInstruction(delegationId, "稍等，我正在查询"); + conversation.appendThinking(delegationId, "本地模拟 Agent：已解析委派参数，准备生成结果"); + conversation.appendCommentary(delegationId, "这是本地模拟结果，未调用任何真实工具"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (RuntimeException e) { + System.out.println("agent worker stopped: " + e.getMessage()); + } + } + + private static JsonObject userTextItem(String text) { + JsonObject part = new JsonObject(); + part.addProperty(FIELD_TYPE, "input_text"); + part.addProperty("text", text); + JsonArray content = new JsonArray(); + content.add(part); + JsonObject item = new JsonObject(); + item.addProperty(FIELD_TYPE, "message"); + item.addProperty("role", "user"); + item.add(FIELD_CONTENT, content); + return item; + } + + private static boolean hasWavHeader(Path pcm) throws Exception { + byte[] magic = new byte[4]; + try (InputStream in = Files.newInputStream(pcm)) { + return in.read(magic) == 4 && "RIFF".equals(new String(magic, "US-ASCII")); + } + } + + private static String text(JsonObject event, String field) { + JsonElement value = event.get(field); + return value != null && value.isJsonPrimitive() ? value.getAsString() : null; + } + + private static class SampleCallback extends QwenLiveCallback { + @Override + public void onOpen() { + System.out.println("websocket opened."); + } + + @Override + public void onEvent(JsonObject message) { + String type = text(message, FIELD_TYPE); + if (type == null) { + System.out.println("event without type: " + message); + return; + } + switch (type) { + case EVENT_SESSION_UPDATED: + sessionUpdated.set(true); + sessionSettled.countDown(); + break; + case EVENT_ERROR: + break; + case EVENT_SESSION_CLOSED: + sessionSettled.countDown(); + break; + case EVENT_SESSION_OUTPUT_AUDIO_DELTA: + String audio = text(message, FIELD_AUDIO); + if (audio == null) { + audio = text(message, FIELD_DELTA); + } + if (audio != null) { + outputAudioBytes.addAndGet(Base64.getDecoder().decode(audio).length); + } + break; + case EVENT_SESSION_INPUT_TRANSCRIPT_DELTA: + append(inputTranscript, text(message, FIELD_DELTA)); + break; + case EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA: + append(outputTranscript, text(message, FIELD_DELTA)); + break; + case EVENT_SESSION_USAGE_UPDATED: + JsonElement usage = message.get(FIELD_USAGE); + if (usage != null && usage.isJsonObject()) { + JsonElement seconds = usage.getAsJsonObject().get(FIELD_SECONDS); + usageSeconds.set(seconds == null ? null : seconds.toString()); + } + break; + case EVENT_SESSION_DELEGATION_CREATED: + if (!delegations.offer(message)) { + System.out.println("delegation queue is full, event dropped: " + message); + } + break; + case EVENT_RESPONSE_EVENT: + System.out.println("backend response event: " + message.get(FIELD_EVENT)); + break; + default: + System.out.println("event: " + type); + } + } + + @Override + public void onClose(int code, String reason) { + System.out.println("websocket closed: " + code + ", " + reason); + sessionSettled.countDown(); + } + + @Override + public void onError(Throwable throwable) { + System.out.println("live error: " + throwable); + sessionSettled.countDown(); + } + } + + private static void append(StringBuilder target, String delta) { + if (delta == null) { + return; + } + synchronized (target) { + target.append(delta); + } + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallback.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallback.java new file mode 100644 index 0000000..5a5f1e7 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallback.java @@ -0,0 +1,18 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import com.google.gson.JsonObject; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public abstract class QwenLiveCallback { + public void onOpen() {} + + public abstract void onEvent(JsonObject message); + + public abstract void onClose(int code, String reason); + + public void onError(Throwable throwable) { + log.warn("Qwen Live callback error.", throwable); + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallbackDispatcher.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallbackDispatcher.java new file mode 100644 index 0000000..2cdcf53 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveCallbackDispatcher.java @@ -0,0 +1,95 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import java.util.ArrayDeque; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; +import java.util.function.Consumer; + +final class QwenLiveCallbackDispatcher { + private final Executor executor; + private final int capacity; + private final Consumer failSession; + private final Consumer reportError; + private final Object lock = new Object(); + private final ArrayDeque queue = new ArrayDeque<>(); + private boolean draining; + private boolean failed; + + QwenLiveCallbackDispatcher( + Executor executor, + int capacity, + Consumer failSession, + Consumer reportError) { + if (capacity <= 0) { + throw new IllegalArgumentException("callbackQueueCapacity must be positive."); + } + this.executor = executor; + this.capacity = capacity; + this.failSession = failSession; + this.reportError = reportError; + } + + void dispatch(Runnable callback) { + if (executor == null) { + callback.run(); + return; + } + boolean schedule = false; + RejectedExecutionException overflow = null; + synchronized (lock) { + if (failed) return; + if (queue.size() >= capacity) { + failed = true; + queue.clear(); + overflow = + new RejectedExecutionException("Qwen Live callback queue is full: " + capacity); + } else { + queue.addLast(callback); + if (!draining) { + draining = true; + schedule = true; + } + } + } + if (overflow != null) { + failSession.accept(overflow); + final Throwable failure = overflow; + synchronized (lock) { + queue.addLast(() -> reportError.accept(failure)); + if (!draining) { + draining = true; + schedule = true; + } + } + } + if (schedule) scheduleDrain(); + } + + private void scheduleDrain() { + try { + executor.execute(this::drain); + } catch (RuntimeException failure) { + synchronized (lock) { + failed = true; + draining = false; + queue.clear(); + } + failSession.accept(failure); + } + } + + private void drain() { + while (true) { + Runnable callback; + synchronized (lock) { + callback = queue.pollFirst(); + if (callback == null) { + draining = false; + return; + } + } + callback.run(); + } + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConnectionParam.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConnectionParam.java new file mode 100644 index 0000000..2f1a782 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConnectionParam.java @@ -0,0 +1,106 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.utils.Constants.DASHSCOPE_WEBSOCKET_OMNI_BASE_URL_ENV; + +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.Executor; +import lombok.Builder; +import lombok.Data; +import lombok.Getter; +import lombok.Singular; +import lombok.experimental.SuperBuilder; +import okhttp3.HttpUrl; + +@Data +@SuperBuilder +public class QwenLiveConnectionParam { + + public static String baseWebsocketApiUrlTemplate = + System.getenv() + .getOrDefault( + DASHSCOPE_WEBSOCKET_OMNI_BASE_URL_ENV, + "wss://{workspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime"); + + private static final String PLACEHOLDER_LOWER_CAMEL = "{workspaceId}"; + + private static final String PLACEHOLDER_UPPER_CAMEL = "{WorkspaceId}"; + + @Getter @lombok.NonNull private String model; + + @Getter private String apikey; + + @Getter private String workspaceId; + + @Getter private String workspace; + + @Singular protected Map headers; + + private String url; + + private Executor callbackExecutor; + + @Builder.Default private int callbackQueueCapacity = 256; + + public Map getHeaders() { + Map res = new HashMap<>(); + for (Map.Entry entry : headers.entrySet()) { + res.put(entry.getKey(), entry.getValue().toString()); + } + return res; + } + + public String getUrl() { + if (isBlank(model)) { + throw new IllegalArgumentException("model must not be blank."); + } + String candidate = url != null ? url : baseWebsocketApiUrlTemplate; + if (isBlank(candidate)) { + throw new IllegalArgumentException("Qwen Live websocket url is not configured."); + } + return appendModelQuery(resolveWorkspacePlaceholder(candidate)); + } + + private String resolveWorkspacePlaceholder(String candidate) { + boolean hasPlaceholder = + candidate.contains(PLACEHOLDER_LOWER_CAMEL) || candidate.contains(PLACEHOLDER_UPPER_CAMEL); + if (!hasPlaceholder) { + return candidate; + } + if (isBlank(workspaceId)) { + throw new IllegalArgumentException( + "workspaceId is required by the websocket url template: " + candidate); + } + return candidate + .replace(PLACEHOLDER_LOWER_CAMEL, workspaceId) + .replace(PLACEHOLDER_UPPER_CAMEL, workspaceId); + } + + // HttpUrl only parses http/https: map ws/wss temporarily, set the model query, then restore. + private String appendModelQuery(String rawUrl) { + String originScheme = null; + String parsableScheme = null; + String parsable = rawUrl; + String lower = rawUrl.toLowerCase(Locale.ROOT); + for (String scheme : new String[] {"wss", "ws", "https", "http"}) { + if (lower.startsWith(scheme + "://")) { + originScheme = scheme; + parsableScheme = "wss".equals(scheme) || "https".equals(scheme) ? "https" : "http"; + parsable = parsableScheme + rawUrl.substring(scheme.length()); + break; + } + } + HttpUrl parsed = HttpUrl.parse(parsable); + if (parsed == null) { + throw new IllegalArgumentException("Invalid websocket url: " + rawUrl); + } + String built = parsed.newBuilder().setQueryParameter("model", model).build().toString(); + return originScheme == null ? built : originScheme + built.substring(parsableScheme.length()); + } + + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConstants.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConstants.java new file mode 100644 index 0000000..2dd6360 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConstants.java @@ -0,0 +1,71 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +public final class QwenLiveConstants { + private QwenLiveConstants() {} + + public static final String EVENT_SESSION_UPDATE = "session.update"; + public static final String EVENT_SESSION_INPUT_AUDIO_APPEND = "session.input_audio.append"; + public static final String EVENT_SESSION_INPUT_AUDIO_MUTE = "session.input_audio.mute"; + public static final String EVENT_SESSION_INPUT_AUDIO_UNMUTE = "session.input_audio.unmute"; + public static final String EVENT_SESSION_INSTRUCTION_APPEND = "session.instruction.append"; + public static final String EVENT_SESSION_THINKING_APPEND = "session.thinking.append"; + public static final String EVENT_SESSION_COMMENTARY_APPEND = "session.commentary.append"; + public static final String EVENT_RESPONSE_ITEM_CREATE = "response.item.create"; + public static final String EVENT_RESPONSE_CREATE = "response.create"; + + public static final String EVENT_ERROR = "error"; + public static final String EVENT_SESSION_CREATED = "session.created"; + public static final String EVENT_SESSION_UPDATED = "session.updated"; + public static final String EVENT_SESSION_INPUT_AUDIO_MUTED = "session.input_audio.muted"; + public static final String EVENT_SESSION_INPUT_AUDIO_UNMUTED = "session.input_audio.unmuted"; + public static final String EVENT_SESSION_OUTPUT_AUDIO_DELTA = "session.output_audio.delta"; + public static final String EVENT_SESSION_INPUT_TRANSCRIPT_DELTA = + "session.input_transcript.delta"; + public static final String EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA = + "session.output_transcript.delta"; + public static final String EVENT_RESPONSE_EVENT = "response.event"; + public static final String EVENT_SESSION_USAGE_UPDATED = "session.usage.updated"; + public static final String EVENT_SESSION_CLOSED = "session.closed"; + public static final String EVENT_SESSION_DELEGATION_CREATED = "session.delegation.created"; + public static final String EVENT_SESSION_INSTRUCTIONS_APPENDED = "session.instructions.appended"; + + public static final String EVENT_SESSION_THINKING_APPENDED = "session.thinking.appended"; + public static final String EVENT_SESSION_COMMENTARY_APPENDED = "session.commentary.appended"; + + public static final String FIELD_TYPE = "type"; + public static final String FIELD_EVENT_ID = "event_id"; + public static final String FIELD_CLIENT_EVENT_ID = "client_event_id"; + public static final String FIELD_SESSION = "session"; + public static final String FIELD_MODEL = "model"; + public static final String FIELD_ID = "id"; + public static final String FIELD_VOICE = "voice"; + public static final String FIELD_INSTRUCTIONS = "instructions"; + public static final String FIELD_MODALITIES = "modalities"; + public static final String FIELD_TOOLS = "tools"; + public static final String FIELD_INPUT_AUDIO_FORMAT = "input_audio_format"; + public static final String FIELD_OUTPUT_AUDIO_FORMAT = "output_audio_format"; + public static final String FIELD_TRANSPORT = "transport"; + public static final String FIELD_DELEGATION = "delegation"; + public static final String FIELD_CONTEXT = "context"; + public static final String FIELD_COMPACTION = "compaction"; + public static final String FIELD_ENABLED = "enabled"; + public static final String FIELD_MODE = "mode"; + public static final String FIELD_RESPONSES = "responses"; + public static final String FIELD_AUDIO = "audio"; + public static final String FIELD_CONTENT = "content"; + public static final String FIELD_ITEM = "item"; + public static final String FIELD_DELEGATION_ID = "delegation_id"; + public static final String FIELD_ERROR = "error"; + public static final String FIELD_REASON = "reason"; + public static final String FIELD_USAGE = "usage"; + public static final String FIELD_CONTEXT_WINDOW = "context_window"; + public static final String FIELD_SECONDS = "seconds"; + public static final String FIELD_USAGE_RATIO = "usage_ratio"; + public static final String FIELD_START_MS = "start_ms"; + public static final String FIELD_END_MS = "end_ms"; + public static final String FIELD_OFFSET_MS = "offset_ms"; + public static final String FIELD_ARGUMENTS = "arguments"; + public static final String FIELD_EVENT = "event"; + public static final String FIELD_DELTA = "delta"; +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContext.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContext.java new file mode 100644 index 0000000..0023795 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContext.java @@ -0,0 +1,26 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.Map; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class QwenLiveContext { + private QwenLiveContextCompaction compaction; + private Map parameters; + + public JsonObject getConfig() { + if (compaction == null) { + throw new IllegalArgumentException("context.compaction is required."); + } + Map fields = QwenLiveSessionConfig.extensionFields(parameters, FIELD_COMPACTION); + fields.put(FIELD_COMPACTION, compaction.getConfig()); + return JsonUtils.toJsonObject(fields).deepCopy(); + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContextCompaction.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContextCompaction.java new file mode 100644 index 0000000..1d67784 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveContextCompaction.java @@ -0,0 +1,27 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.Map; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class QwenLiveContextCompaction { + private Boolean enabled; + private String mode; + private Map parameters; + + public JsonObject getConfig() { + QwenLiveSessionConfig.requireText(mode, "context.compaction.mode"); + Map fields = + QwenLiveSessionConfig.extensionFields(parameters, FIELD_ENABLED, FIELD_MODE); + QwenLiveSessionConfig.putIfSet(fields, FIELD_ENABLED, enabled); + fields.put(FIELD_MODE, mode); + return JsonUtils.toJsonObject(fields).deepCopy(); + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConversation.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConversation.java new file mode 100644 index 0000000..0b552fe --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveConversation.java @@ -0,0 +1,477 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.common.Status; +import com.alibaba.dashscope.exception.ApiException; +import com.alibaba.dashscope.exception.NoApiKeyException; +import com.alibaba.dashscope.protocol.DashScopeHeaders; +import com.alibaba.dashscope.protocol.okhttp.OkHttpClientFactory; +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.internal.Streams; +import com.google.gson.stream.JsonReader; +import com.google.gson.stream.JsonToken; +import java.io.IOException; +import java.io.StringReader; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Base64; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import lombok.extern.slf4j.Slf4j; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.WebSocket; +import okhttp3.WebSocketListener; + +@Slf4j +public class QwenLiveConversation { + private static final Gson EVENT_GSON = JsonUtils.gson.newBuilder().serializeNulls().create(); + private final QwenLiveConnectionParam param; + private final QwenLiveCallback callback; + private final QwenLiveCallbackDispatcher callbackDispatcher; + private final WebSocket.Factory factory; + private final int connectTimeoutSeconds; + private final Object lifecycleLock = new Object(); + private final AtomicBoolean connectAttempted = new AtomicBoolean(false); + private final AtomicBoolean isOpen = new AtomicBoolean(false); + private final AtomicBoolean isClosed = new AtomicBoolean(false); + private final AtomicBoolean sessionClosedReceived = new AtomicBoolean(false); + private final AtomicReference connectLatch = new AtomicReference<>(); + private final AtomicReference sessionClosedLatch = new AtomicReference<>(); + private volatile WebSocket webSocketClient; + private volatile Throwable connectFailure; + private volatile String sessionId; + private volatile String closeReason; + private boolean closeRequested; + private boolean socketCancelled; + + public QwenLiveConversation(QwenLiveConnectionParam param, QwenLiveCallback callback) { + this(param, callback, OkHttpClientFactory.getOkHttpClient(), 60); + } + + QwenLiveConversation( + QwenLiveConnectionParam param, + QwenLiveCallback callback, + WebSocket.Factory factory, + int connectTimeoutSeconds) { + this.param = Objects.requireNonNull(param, "param"); + this.callback = Objects.requireNonNull(callback, "callback"); + this.factory = Objects.requireNonNull(factory, "factory"); + if (connectTimeoutSeconds <= 0) { + throw new IllegalArgumentException("connectTimeoutSeconds must be positive."); + } + this.connectTimeoutSeconds = connectTimeoutSeconds; + this.callbackDispatcher = + new QwenLiveCallbackDispatcher( + param.getCallbackExecutor(), + param.getCallbackQueueCapacity(), + this::failCallbackDispatch, + this::notifyError); + } + + public void connect() throws NoApiKeyException, InterruptedException { + synchronized (lifecycleLock) { + if (isClosed.get() || !connectAttempted.compareAndSet(false, true)) { + throw new IllegalStateException("Qwen Live conversation cannot connect again."); + } + connectLatch.set(new CountDownLatch(1)); + sessionClosedLatch.set(new CountDownLatch(1)); + } + try { + Request request = buildConnectionRequest(); + WebSocket created = + Objects.requireNonNull(factory.newWebSocket(request, new Listener()), "socket"); + boolean terminal; + synchronized (lifecycleLock) { + terminal = isClosed.get() || connectFailure != null; + if (!terminal) { + webSocketClient = created; + } + } + if (terminal) cancelSocket(created); + if (!connectLatch.get().await(connectTimeoutSeconds, TimeUnit.SECONDS)) { + cancel(); + throw new IllegalStateException("Timed out connecting Qwen Live websocket."); + } + synchronized (lifecycleLock) { + if (connectFailure != null) { + throw new IllegalStateException("Failed connecting Qwen Live websocket.", connectFailure); + } + checkStatus(); + } + } catch (InterruptedException e) { + cancel(); + Thread.currentThread().interrupt(); + throw e; + } catch (NoApiKeyException | RuntimeException e) { + synchronized (lifecycleLock) { + if (connectFailure == null) connectFailure = e; + markClosed(); + } + cancel(); + throw e; + } + } + + private Request buildConnectionRequest() throws NoApiKeyException { + Request.Builder builder = new Request.Builder().url(param.getUrl()); + Map headers = + DashScopeHeaders.buildWebSocketHeaders( + param.getApikey(), false, param.getWorkspace(), param.getHeaders(), null, "audio"); + for (Map.Entry header : headers.entrySet()) { + builder.header(header.getKey(), header.getValue()); + } + return builder.build(); + } + + public void checkStatus() { + if (!isOpen.get() || isClosed.get() || webSocketClient == null) { + throw new IllegalStateException("Qwen Live websocket is not open."); + } + } + + public String updateSession(QwenLiveSessionConfig config) { + requireValue(config, "config"); + JsonObject payload = new JsonObject(); + payload.add(FIELD_SESSION, config.getConfig()); + return sendEvent(EVENT_SESSION_UPDATE, payload, true); + } + + public String appendAudio(String audioBase64) { + requireValue(audioBase64, "audioBase64"); + JsonObject payload = new JsonObject(); + payload.addProperty(FIELD_AUDIO, audioBase64); + return sendEvent(EVENT_SESSION_INPUT_AUDIO_APPEND, payload, false); + } + + public String appendAudioBytes(byte[] audio) { + requireValue(audio, "audio"); + return appendAudio(Base64.getEncoder().encodeToString(audio)); + } + + public String appendAudioBytes(byte[] audio, int offset, int length) { + requireValue(audio, "audio"); + if (offset < 0 || length < 0 || offset > audio.length - length) { + throw new IllegalArgumentException("Audio slice is out of bounds."); + } + return appendAudio( + Base64.getEncoder().encodeToString(Arrays.copyOfRange(audio, offset, offset + length))); + } + + public String muteInputAudio() { + return sendEvent(EVENT_SESSION_INPUT_AUDIO_MUTE, new JsonObject(), true); + } + + public String unmuteInputAudio() { + return sendEvent(EVENT_SESSION_INPUT_AUDIO_UNMUTE, new JsonObject(), true); + } + + public String appendInstruction(String delegationId, String content) { + return sendDelegationContent(EVENT_SESSION_INSTRUCTION_APPEND, delegationId, content); + } + + public String appendThinking(String delegationId, String content) { + return sendDelegationContent(EVENT_SESSION_THINKING_APPEND, delegationId, content); + } + + public String appendCommentary(String delegationId, String content) { + return sendDelegationContent(EVENT_SESSION_COMMENTARY_APPEND, delegationId, content); + } + + private String sendDelegationContent(String type, String delegationId, String content) { + QwenLiveSessionConfig.requireText(delegationId, FIELD_DELEGATION_ID); + requireValue(content, FIELD_CONTENT); + JsonObject payload = new JsonObject(); + payload.addProperty(FIELD_DELEGATION_ID, delegationId); + payload.addProperty(FIELD_CONTENT, content); + return sendEvent(type, payload, true); + } + + public String createResponseItem(JsonObject item) { + requireValue(item, FIELD_ITEM); + JsonObject payload = new JsonObject(); + payload.add(FIELD_ITEM, item.deepCopy()); + return sendEvent(EVENT_RESPONSE_ITEM_CREATE, payload, true); + } + + public String createResponse() { + return sendEvent(EVENT_RESPONSE_CREATE, new JsonObject(), true); + } + + public String sendRaw(String rawJson) { + JsonObject payload = parseObject(rawJson); + String type = stringField(payload, FIELD_TYPE); + QwenLiveSessionConfig.requireText(type, FIELD_TYPE); + return sendEvent(type, payload, !EVENT_SESSION_INPUT_AUDIO_APPEND.equals(type)); + } + + private String sendEvent(String type, JsonObject payload, boolean enableLog) { + JsonObject event = payload.deepCopy(); + String eventId = "event_" + UUID.randomUUID().toString().replace("-", ""); + event.addProperty(FIELD_TYPE, type); + event.addProperty(FIELD_EVENT_ID, eventId); + String message = EVENT_GSON.toJson(event); + synchronized (lifecycleLock) { + checkStatus(); + WebSocket socket = webSocketClient; + if (socket == null || !socket.send(message)) { + throw new IllegalStateException("Failed to enqueue Qwen Live event: " + eventId); + } + } + if (enableLog) { + log.debug("Qwen Live sent: {}", message); + } else { + log.debug("Qwen Live audio event: {}, message length: {}", eventId, message.length()); + } + return eventId; + } + + private static void requireValue(Object value, String field) { + if (value == null) throw new IllegalArgumentException(field + " must not be null."); + } + + private static JsonObject parseObject(String json) { + requireValue(json, "json"); + try (JsonReader reader = new JsonReader(new StringReader(json))) { + reader.setLenient(false); + JsonElement parsed = Streams.parse(reader); + if (!parsed.isJsonObject() || reader.peek() != JsonToken.END_DOCUMENT) { + throw new IllegalArgumentException("Expected a single JSON object."); + } + return parsed.getAsJsonObject(); + } catch (IOException | com.google.gson.JsonParseException e) { + throw new IllegalArgumentException("Invalid JSON object.", e); + } + } + + private static String stringField(JsonObject object, String field) { + JsonElement value = object.get(field); + return value != null && value.isJsonPrimitive() && value.getAsJsonPrimitive().isString() + ? value.getAsString() + : null; + } + + public void close() { + close(1000, "bye"); + } + + public void close(int code, String reason) { + validateClose(code, reason); + WebSocket socket; + boolean wasOpen; + synchronized (lifecycleLock) { + if (closeRequested || socketCancelled) { + return; + } + closeRequested = true; + wasOpen = isOpen.get(); + socket = webSocketClient; + markClosing(); + } + if (socket == null || !wasOpen || !socket.close(code, reason)) { + cancel(); + } + } + + public void cancel() { + WebSocket socket; + synchronized (lifecycleLock) { + markClosed(); + socket = webSocketClient; + } + cancelSocket(socket); + } + + private void cancelSocket(WebSocket socket) { + if (socket == null) return; + synchronized (lifecycleLock) { + if (socketCancelled) return; + socketCancelled = true; + } + socket.cancel(); + } + + private static void validateClose(int code, String reason) { + if (code < 1000 + || code >= 5000 + || (code >= 1004 && code <= 1006) + || (code >= 1015 && code < 3000)) { + throw new IllegalArgumentException("Invalid websocket close code: " + code); + } + if (reason != null && reason.getBytes(StandardCharsets.UTF_8).length > 123) { + throw new IllegalArgumentException("Websocket close reason exceeds 123 UTF-8 bytes."); + } + } + + public boolean awaitSessionClosed(int timeoutSeconds) throws InterruptedException { + if (timeoutSeconds < 0) + throw new IllegalArgumentException("timeoutSeconds must not be negative."); + if (sessionClosedReceived.get()) return true; + CountDownLatch latch = sessionClosedLatch.get(); + if (latch == null) return false; + try { + latch.await(timeoutSeconds, TimeUnit.SECONDS); + return sessionClosedReceived.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw e; + } + } + + public String getSessionId() { + return sessionId; + } + + public String getCloseReason() { + return closeReason; + } + + private void markClosing() { + isClosed.set(true); + isOpen.set(false); + CountDownLatch connecting = connectLatch.get(); + if (connecting != null) connecting.countDown(); + } + + private void markClosed() { + markClosing(); + CountDownLatch closing = sessionClosedLatch.get(); + if (closing != null) closing.countDown(); + } + + private void dispatchCallback(Runnable action) { + callbackDispatcher.dispatch( + () -> { + try { + action.run(); + } catch (Throwable failure) { + notifyError(failure); + } + }); + } + + private void reportError(Throwable failure) { + callbackDispatcher.dispatch(() -> notifyError(failure)); + } + + private void notifyError(Throwable failure) { + try { + callback.onError(failure); + } catch (Throwable callbackFailure) { + log.warn("Qwen Live onError callback failed.", callbackFailure); + } + } + + private void failCallbackDispatch(Throwable failure) { + log.warn("Qwen Live callback dispatch failed; cancelling session.", failure); + synchronized (lifecycleLock) { + if (connectFailure == null) connectFailure = failure; + } + cancel(); + } + + private static ApiException serverError(JsonObject event) { + JsonElement detail = event.get(FIELD_ERROR); + JsonObject error = + detail != null && detail.isJsonObject() ? detail.getAsJsonObject() : new JsonObject(); + String code = stringField(error, "code"); + if (code == null) code = stringField(error, FIELD_TYPE); + String message = stringField(error, "message"); + return new ApiException( + Status.builder() + .statusCode(-1) + .code(code) + .message(message != null ? message : "Qwen Live server error.") + .requestId(stringField(event, "request_id")) + .build()); + } + + private class Listener extends WebSocketListener { + @Override + public void onOpen(WebSocket socket, Response response) { + boolean terminal; + synchronized (lifecycleLock) { + terminal = isClosed.get(); + if (!terminal) { + webSocketClient = socket; + isOpen.set(true); + CountDownLatch latch = connectLatch.get(); + if (latch != null) latch.countDown(); + } + } + if (terminal) { + cancelSocket(socket); + return; + } + dispatchCallback(callback::onOpen); + } + + @Override + public void onMessage(WebSocket socket, String text) { + JsonObject event; + try { + event = parseObject(text); + } catch (RuntimeException failure) { + reportError(failure); + return; + } + String type = stringField(event, FIELD_TYPE); + synchronized (lifecycleLock) { + if (EVENT_SESSION_CREATED.equals(type)) { + JsonElement session = event.get(FIELD_SESSION); + if (session != null && session.isJsonObject()) { + String id = stringField(session.getAsJsonObject(), FIELD_ID); + if (id != null) sessionId = id; + } + } else if (EVENT_SESSION_CLOSED.equals(type)) { + closeReason = stringField(event, FIELD_REASON); + sessionClosedReceived.set(true); + markClosed(); + } + } + ApiException failure = EVENT_ERROR.equals(type) ? serverError(event) : null; + dispatchCallback( + () -> { + if (failure != null) notifyError(failure); + callback.onEvent(event); + }); + } + + @Override + public void onClosing(WebSocket socket, int code, String reason) { + synchronized (lifecycleLock) { + markClosed(); + } + if (!socket.close(code, reason)) cancelSocket(socket); + } + + @Override + public void onClosed(WebSocket socket, int code, String reason) { + synchronized (lifecycleLock) { + markClosed(); + } + dispatchCallback(() -> callback.onClose(code, reason)); + } + + @Override + public void onFailure(WebSocket socket, Throwable failure, Response response) { + synchronized (lifecycleLock) { + if (webSocketClient == null) webSocketClient = socket; + if (connectFailure == null) connectFailure = failure; + markClosed(); + } + cancelSocket(socket); + reportError(failure); + } + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegation.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegation.java new file mode 100644 index 0000000..9c1d5c7 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegation.java @@ -0,0 +1,35 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.LinkedHashMap; +import java.util.Map; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class QwenLiveDelegation { + private QwenLiveDelegationType type; + private QwenLiveResponsesConfig responses; + + public JsonObject getConfig() { + if (type == null) { + throw new IllegalArgumentException("delegation.type is required."); + } + Map fields = new LinkedHashMap<>(); + fields.put(FIELD_TYPE, type); + if (type == QwenLiveDelegationType.RESPONSES) { + if (responses == null) { + throw new IllegalArgumentException("delegation.responses is required in responses mode."); + } + fields.put(FIELD_RESPONSES, responses.getConfig()); + } else if (responses != null) { + throw new IllegalArgumentException("only responses delegation may carry responses."); + } + return JsonUtils.toJsonObject(fields).deepCopy(); + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegationType.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegationType.java new file mode 100644 index 0000000..2927eba --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveDelegationType.java @@ -0,0 +1,13 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import com.google.gson.annotations.SerializedName; + +public enum QwenLiveDelegationType { + @SerializedName("none") + NONE, + @SerializedName("client") + CLIENT, + @SerializedName("responses") + RESPONSES +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveResponsesConfig.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveResponsesConfig.java new file mode 100644 index 0000000..bf5a35d --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveResponsesConfig.java @@ -0,0 +1,31 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.List; +import java.util.Map; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class QwenLiveResponsesConfig { + private String model; + private String instructions; + private List tools; + private Map parameters; + + public JsonObject getConfig() { + QwenLiveSessionConfig.requireText(model, "responses.model"); + QwenLiveSessionConfig.requireText(instructions, "responses.instructions"); + Map fields = + QwenLiveSessionConfig.extensionFields(parameters, FIELD_MODEL, FIELD_INSTRUCTIONS, FIELD_TOOLS); + fields.put(FIELD_MODEL, model); + fields.put(FIELD_INSTRUCTIONS, instructions); + QwenLiveSessionConfig.putIfSet(fields, FIELD_TOOLS, tools); + return JsonUtils.toJsonObject(fields).deepCopy(); + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveSessionConfig.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveSessionConfig.java new file mode 100644 index 0000000..5a6e059 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveSessionConfig.java @@ -0,0 +1,86 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; + +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class QwenLiveSessionConfig { + private String model; + private String instructions; + private String voice; + private String inputAudioFormat; + private String outputAudioFormat; + private List modalities; + private List tools; + private QwenLiveDelegation delegation; + private QwenLiveTransport transport; + private QwenLiveContext context; + private Map parameters; + + public JsonObject getConfig() { + requireText(model, FIELD_MODEL); + if (delegation == null) { + throw new IllegalArgumentException("session.delegation is required."); + } + Map fields = + extensionFields( + parameters, + FIELD_MODEL, + FIELD_INSTRUCTIONS, + FIELD_VOICE, + FIELD_INPUT_AUDIO_FORMAT, + FIELD_OUTPUT_AUDIO_FORMAT, + FIELD_MODALITIES, + FIELD_TOOLS, + FIELD_DELEGATION, + FIELD_TRANSPORT, + FIELD_CONTEXT); + fields.put(FIELD_MODEL, model); + putIfSet(fields, FIELD_INSTRUCTIONS, instructions); + putIfSet(fields, FIELD_VOICE, voice); + putIfSet(fields, FIELD_INPUT_AUDIO_FORMAT, inputAudioFormat); + putIfSet(fields, FIELD_OUTPUT_AUDIO_FORMAT, outputAudioFormat); + putIfSet(fields, FIELD_MODALITIES, modalities); + putIfSet(fields, FIELD_TOOLS, tools); + if (context != null) { + fields.put(FIELD_CONTEXT, context.getConfig()); + } + fields.put(FIELD_DELEGATION, delegation.getConfig()); + if (transport != null) { + fields.put(FIELD_TRANSPORT, transport.getConfig()); + } + return JsonUtils.toJsonObject(fields).deepCopy(); + } + + static Map extensionFields(Map parameters, String... reserved) { + Map fields = new LinkedHashMap<>(); + if (parameters != null) { + fields.putAll(parameters); + } + for (String key : reserved) { + fields.remove(key); + } + return fields; + } + + static void putIfSet(Map fields, String key, Object value) { + if (value != null) { + fields.put(key, value); + } + } + + static void requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(field + " must not be blank."); + } + } +} diff --git a/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveTransport.java b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveTransport.java new file mode 100644 index 0000000..13aaa98 --- /dev/null +++ b/src/main/java/com/alibaba/dashscope/audio/live/QwenLiveTransport.java @@ -0,0 +1,24 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.FIELD_TYPE; + +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.Map; +import lombok.Data; +import lombok.experimental.SuperBuilder; + +@Data +@SuperBuilder +public class QwenLiveTransport { + private String type; + private Map parameters; + + public JsonObject getConfig() { + QwenLiveSessionConfig.requireText(type, "transport.type"); + Map fields = QwenLiveSessionConfig.extensionFields(parameters, FIELD_TYPE); + fields.put(FIELD_TYPE, type); + return JsonUtils.toJsonObject(fields).deepCopy(); + } +} diff --git a/src/test/java/com/alibaba/dashscope/TestQwenLiveConnectionParam.java b/src/test/java/com/alibaba/dashscope/TestQwenLiveConnectionParam.java new file mode 100644 index 0000000..f6ed201 --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/TestQwenLiveConnectionParam.java @@ -0,0 +1,182 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.alibaba.dashscope.audio.live.QwenLiveConnectionParam; +import java.util.HashMap; +import java.util.Map; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TestQwenLiveConnectionParam { + + private static final String DEFAULT_TEMPLATE = + "wss://{workspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime"; + + private String savedTemplate; + + @BeforeEach + void setUp() { + savedTemplate = QwenLiveConnectionParam.baseWebsocketApiUrlTemplate; + QwenLiveConnectionParam.baseWebsocketApiUrlTemplate = DEFAULT_TEMPLATE; + } + + @AfterEach + void tearDown() { + QwenLiveConnectionParam.baseWebsocketApiUrlTemplate = savedTemplate; + } + + @Test + void workspaceRoutesToNewGateway() { + QwenLiveConnectionParam p = + QwenLiveConnectionParam.builder().model("qwen-live").workspaceId("workspace-test").build(); + assertEquals( + "wss://workspace-test.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen-live", + p.getUrl()); + assertNull(p.getWorkspace()); + } + + @Test + void templateNeedsWorkspaceId() { + assertThrows( + IllegalArgumentException.class, + () -> QwenLiveConnectionParam.builder().model("qwen-live").build().getUrl()); + } + + @Test + void explicitUrlDoesNotNeedWorkspaceId() { + assertEquals( + "ws://localhost:8080/realtime?model=qwen-live", + QwenLiveConnectionParam.builder() + .model("qwen-live") + .url("ws://localhost:8080/realtime") + .build() + .getUrl()); + } + + @Test + void documentCasePlaceholderIsAlsoReplaced() { + QwenLiveConnectionParam.baseWebsocketApiUrlTemplate = + "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime"; + assertEquals( + "wss://ws-1.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen-live", + QwenLiveConnectionParam.builder().model("qwen-live").workspaceId("ws-1").build().getUrl()); + } + + @Test + void placeholderInExplicitUrlIsReplaced() { + assertEquals( + "wss://ws-2.example.com/realtime?model=qwen-live", + QwenLiveConnectionParam.builder() + .model("qwen-live") + .workspaceId("ws-2") + .url("wss://{workspaceId}.example.com/realtime") + .build() + .getUrl()); + } + + @Test + void fixedTemplateDoesNotNeedWorkspaceId() { + QwenLiveConnectionParam.baseWebsocketApiUrlTemplate = "wss://fixed.example.com/api-ws/v1/realtime"; + assertEquals( + "wss://fixed.example.com/api-ws/v1/realtime?model=qwen-live", + QwenLiveConnectionParam.builder().model("qwen-live").build().getUrl()); + } + + @Test + void existingQueryIsPreserved() { + assertEquals( + "ws://localhost:8080/realtime?foo=bar&model=qwen-live", + QwenLiveConnectionParam.builder() + .model("qwen-live") + .url("ws://localhost:8080/realtime?foo=bar") + .build() + .getUrl()); + } + + @Test + void existingModelQueryIsReplacedOnce() { + String actual = + QwenLiveConnectionParam.builder() + .model("qwen-live") + .url("ws://localhost:8080/realtime?model=stale") + .build() + .getUrl(); + assertEquals("ws://localhost:8080/realtime?model=qwen-live", actual); + } + + @Test + void modelSpecialCharactersAreEncoded() { + String actual = + QwenLiveConnectionParam.builder() + .model("qwen live&x") + .url("ws://localhost:8080/realtime") + .build() + .getUrl(); + assertEquals("ws://localhost:8080/realtime?model=qwen%20live%26x", actual); + } + + @Test + void blankModelIsRejected() { + assertThrows( + IllegalArgumentException.class, + () -> QwenLiveConnectionParam.builder().model(" ").workspaceId("ws-1").build().getUrl()); + } + + @Test + void illegalSchemeIsRejected() { + for (String bad : + new String[] {"ftp://host/realtime", "host/realtime", "wsss://host/realtime"}) { + assertThrows( + IllegalArgumentException.class, + () -> QwenLiveConnectionParam.builder().model("qwen-live").url(bad).build().getUrl(), + "should reject url: [" + bad + "]"); + } + } + + @Test + void httpSchemeIsAllowedForLocalServer() { + assertEquals( + "http://127.0.0.1:9000/realtime?model=qwen-live", + QwenLiveConnectionParam.builder() + .model("qwen-live") + .url("http://127.0.0.1:9000/realtime") + .build() + .getUrl()); + } + + @Test + void workspaceHeaderIsIndependentFromWorkspaceId() { + QwenLiveConnectionParam p = + QwenLiveConnectionParam.builder() + .model("qwen-live") + .workspaceId("ws-host") + .workspace("llm-xxxx") + .build(); + assertTrue(p.getUrl().startsWith("wss://ws-host.")); + assertEquals("llm-xxxx", p.getWorkspace()); + } + + @Test + void headersAreConvertedToStringAndInputIsNotMutated() { + Map source = new HashMap<>(); + source.put("X-Test", 1); + QwenLiveConnectionParam p = + QwenLiveConnectionParam.builder().model("qwen-live").workspaceId("ws-1").headers(source).build(); + Map headers = p.getHeaders(); + assertEquals("1", headers.get("X-Test")); + assertEquals(1, source.size()); + assertEquals(1, source.get("X-Test")); + } + + @Test + void emptyHeadersAreSafe() { + QwenLiveConnectionParam p = QwenLiveConnectionParam.builder().model("qwen-live").workspaceId("ws-1").build(); + assertTrue(p.getHeaders().isEmpty()); + } +} diff --git a/src/test/java/com/alibaba/dashscope/TestQwenLiveSessionConfig.java b/src/test/java/com/alibaba/dashscope/TestQwenLiveSessionConfig.java new file mode 100644 index 0000000..ef450ed --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/TestQwenLiveSessionConfig.java @@ -0,0 +1,295 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope; + +import static org.junit.jupiter.api.Assertions.*; + +import com.alibaba.dashscope.audio.live.*; +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; + +public class TestQwenLiveSessionConfig { + private QwenLiveDelegation client() { + return QwenLiveDelegation.builder().type(QwenLiveDelegationType.CLIENT).build(); + } + + @Test + void clientConfigOmitsUnsetKeys() { + assertEquals( + JsonUtils.parse("{\"model\":\"front\",\"delegation\":{\"type\":\"client\"}}"), + QwenLiveSessionConfig.builder().model("front").delegation(client()).build().getConfig()); + } + + @Test + void responsesAreNested() { + JsonObject actual = + QwenLiveSessionConfig.builder() + .model("front") + .delegation( + QwenLiveDelegation.builder() + .type(QwenLiveDelegationType.RESPONSES) + .responses( + QwenLiveResponsesConfig.builder() + .model("backend-test") + .instructions("处理任务") + .tools(Collections.emptyList()) + .build()) + .build()) + .build() + .getConfig(); + assertEquals( + JsonUtils.parse( + "{\"model\":\"front\",\"delegation\":{\"type\":\"responses\",\"responses\":{" + + "\"model\":\"backend-test\",\"instructions\":\"处理任务\",\"tools\":[]}}}"), + actual); + assertFalse(actual.has("turn_detection")); + assertFalse(actual.has("delegate")); + } + + @Test + void invalidDelegationCombinationsAreRejected() { + assertThrows( + IllegalArgumentException.class, () -> QwenLiveSessionConfig.builder().build().getConfig()); + assertThrows( + IllegalArgumentException.class, () -> QwenLiveDelegation.builder().build().getConfig()); + assertThrows( + IllegalArgumentException.class, + () -> + QwenLiveDelegation.builder() + .type(QwenLiveDelegationType.RESPONSES) + .build() + .getConfig()); + assertThrows( + IllegalArgumentException.class, + () -> + QwenLiveDelegation.builder() + .type(QwenLiveDelegationType.CLIENT) + .responses( + QwenLiveResponsesConfig.builder().model("backend").instructions("任务").build()) + .build() + .getConfig()); + } + + @Test + void backendModelAndInstructionsAreRequired() { + for (String bad : new String[] {null, "", " "}) { + assertThrows( + IllegalArgumentException.class, + () -> + QwenLiveResponsesConfig.builder().model(bad).instructions("任务").build().getConfig()); + assertThrows( + IllegalArgumentException.class, + () -> + QwenLiveResponsesConfig.builder() + .model("backend") + .instructions(bad) + .build() + .getConfig()); + } + } + + @Test + void typedFieldsAndToolsHaveExactWireNames() { + JsonObject tool = + JsonUtils.parse( + "{\"type\":\"function\",\"function\":{\"name\":\"weather\"," + + "\"parameters\":{\"type\":\"object\",\"properties\":{\"city\":{\"type\":\"string\"}}}}}"); + JsonObject original = tool.deepCopy(); + JsonObject actual = + QwenLiveSessionConfig.builder() + .model("front") + .instructions("前台") + .voice("voice") + .inputAudioFormat("pcm16") + .outputAudioFormat("pcm24") + .modalities(Arrays.asList("audio", "text")) + .tools(Collections.singletonList(tool)) + .delegation(client()) + .build() + .getConfig(); + assertEquals("front", actual.get("model").getAsString()); + assertEquals("前台", actual.get("instructions").getAsString()); + assertEquals("voice", actual.get("voice").getAsString()); + assertEquals("pcm16", actual.get("input_audio_format").getAsString()); + assertEquals("pcm24", actual.get("output_audio_format").getAsString()); + assertEquals(JsonUtils.parseString("[\"audio\",\"text\"]"), actual.get("modalities")); + assertEquals(tool, actual.getAsJsonArray("tools").get(0)); + actual.getAsJsonArray("tools").get(0).getAsJsonObject().addProperty("changed", true); + assertEquals(original, tool); + assertFalse(actual.has("inputAudioFormat")); + } + + @Test + void topLevelReservedKeysCannotBeInjectedEvenWhenUnset() { + Map extra = new LinkedHashMap<>(); + for (String key : + new String[] { + "model", + "instructions", + "voice", + "input_audio_format", + "output_audio_format", + "modalities", + "tools", + "delegation", + "transport", + "context" + }) { + extra.put(key, "injected"); + } + extra.put("turn_detection", JsonUtils.parse("{\"type\":\"server_vad\"}")); + extra.put("input_audio_transcription", JsonUtils.parse("{\"model\":\"asr\"}")); + extra.put("custom", "扩展"); + Map original = new LinkedHashMap<>(extra); + JsonObject actual = + QwenLiveSessionConfig.builder() + .model("front") + .delegation(client()) + .parameters(extra) + .build() + .getConfig(); + assertEquals(5, actual.size()); + assertEquals("front", actual.get("model").getAsString()); + assertEquals("client", actual.getAsJsonObject("delegation").get("type").getAsString()); + assertEquals("扩展", actual.get("custom").getAsString()); + assertTrue(actual.has("turn_detection")); + assertTrue(actual.has("input_audio_transcription")); + assertFalse(actual.has("context")); + assertFalse(actual.has("parameters")); + assertEquals(original, extra); + } + + @Test + void backendExtensionsCannotOverrideReservedKeys() { + Map extra = new LinkedHashMap<>(); + extra.put("model", "bad"); + extra.put("instructions", "bad"); + extra.put("tools", "bad"); + extra.put("temperature", 0.4); + JsonObject actual = + QwenLiveResponsesConfig.builder() + .model("backend") + .instructions("任务") + .parameters(extra) + .build() + .getConfig(); + assertEquals("backend", actual.get("model").getAsString()); + assertEquals("任务", actual.get("instructions").getAsString()); + assertFalse(actual.has("tools")); + assertEquals(0.4, actual.get("temperature").getAsDouble()); + assertFalse(actual.has("parameters")); + assertEquals(4, extra.size()); + } + + @Test + void transportIsAnOptionalFlattenedConfiguration() { + Map extra = new LinkedHashMap<>(); + extra.put("type", "bad"); + extra.put("custom", true); + JsonObject actual = + QwenLiveSessionConfig.builder() + .model("front") + .delegation(client()) + .transport(QwenLiveTransport.builder().type("aoq").parameters(extra).build()) + .build() + .getConfig(); + assertEquals(JsonUtils.parse("{\"type\":\"aoq\",\"custom\":true}"), actual.get("transport")); + assertEquals( + "websocket", + QwenLiveTransport.builder() + .type("websocket") + .build() + .getConfig() + .get("type") + .getAsString()); + assertThrows( + IllegalArgumentException.class, () -> QwenLiveTransport.builder().build().getConfig()); + assertThrows( + IllegalArgumentException.class, + () -> QwenLiveTransport.builder().type(" ").build().getConfig()); + } + + @Test + void settersCanUpdateConfiguration() { + QwenLiveSessionConfig config = QwenLiveSessionConfig.builder().model("front").delegation(client()).build(); + config.setVoice("new-voice"); + config.setTools(Collections.emptyList()); + assertEquals("new-voice", config.getConfig().get("voice").getAsString()); + assertEquals(0, config.getConfig().getAsJsonArray("tools").size()); + QwenLiveResponsesConfig backend = + QwenLiveResponsesConfig.builder().model("b").instructions("任务").build(); + backend.setModel("new-backend"); + QwenLiveDelegation d = client(); + d.setType(QwenLiveDelegationType.RESPONSES); + d.setResponses(backend); + config.setDelegation(d); + QwenLiveTransport transport = QwenLiveTransport.builder().type("aoq").build(); + transport.setType("websocket"); + config.setTransport(transport); + assertEquals( + "new-backend", + config + .getConfig() + .getAsJsonObject("delegation") + .getAsJsonObject("responses") + .get("model") + .getAsString()); + assertEquals( + "websocket", config.getConfig().getAsJsonObject("transport").get("type").getAsString()); + } + + @Test + void modelIsRequired() { + for (String bad : new String[] {null, "", " "}) { + assertThrows( + IllegalArgumentException.class, + () -> QwenLiveSessionConfig.builder().model(bad).delegation(client()).build().getConfig()); + } + } + + @Test + void delegationTypeNoneIsSupported() { + assertEquals( + JsonUtils.parse("{\"model\":\"front\",\"delegation\":{\"type\":\"none\"}}"), + QwenLiveSessionConfig.builder() + .model("front") + .delegation(QwenLiveDelegation.builder().type(QwenLiveDelegationType.NONE).build()) + .build() + .getConfig()); + } + + @Test + void contextCompactionIsTypedAndNested() { + JsonObject actual = + QwenLiveSessionConfig.builder() + .model("front") + .delegation(client()) + .context( + QwenLiveContext.builder() + .compaction( + QwenLiveContextCompaction.builder().enabled(false).mode("local").build()) + .build()) + .build() + .getConfig(); + assertEquals( + JsonUtils.parse( + "{\"model\":\"front\",\"context\":{\"compaction\":{\"enabled\":false,\"mode\":\"local\"}}," + + "\"delegation\":{\"type\":\"client\"}}"), + actual); + } + + @Test + void contextRequiresCompactionWithMode() { + assertThrows( + IllegalArgumentException.class, () -> QwenLiveContext.builder().build().getConfig()); + for (String bad : new String[] {null, "", " "}) { + assertThrows( + IllegalArgumentException.class, + () -> QwenLiveContextCompaction.builder().mode(bad).build().getConfig()); + } + } +} diff --git a/src/test/java/com/alibaba/dashscope/audio/live/QwenLiveQuickStartChecks.java b/src/test/java/com/alibaba/dashscope/audio/live/QwenLiveQuickStartChecks.java new file mode 100644 index 0000000..ea45af3 --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/audio/live/QwenLiveQuickStartChecks.java @@ -0,0 +1,406 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import com.alibaba.dashscope.utils.JsonUtils; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.channels.ClosedByInterruptException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.Base64; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import okhttp3.WebSocket; +import okhttp3.WebSocketListener; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; + +public final class QwenLiveQuickStartChecks { + public static void main(String[] args) throws Exception { + try { + savesExactAudioFragments(); + silenceWithoutAudioTimesOut(); + newAudioRestartsIdleWindow(); + earlyAudioStillWaitsAfterInputEnds(); + errorsUnblockWaitingAndPreserveTheirCause(); + fileWriteFailureIsNotReportedAsSuccess(); + sessionUpdateMustBeConfirmed(); + malformedAudioIsNotReportedAsSuccess(); + doesNotOverwriteOutputAndRejectsBadWaitSettings(); + fileToReplyLoopWaitsForDelayedAudio(); + standaloneCliExitsAfterSuccessAndFailure(); + invalidCliArgumentsReturnFailure(); + System.out.println("QUICK_START_CHECKS_OK: 15 cases"); + } finally { + okhttp3.OkHttpClient client = com.alibaba.dashscope.protocol.okhttp.OkHttpClientFactory.getOkHttpClient(); + client.dispatcher().executorService().shutdownNow(); + client.connectionPool().evictAll(); + } + } + + private static final class Scenario implements AutoCloseable { + final Path directory = Files.createTempDirectory(Paths.get("target"), "quickstart-scenario-"); + final Path input = directory.resolve("input.pcm"); + final Path output = directory.resolve("reply.pcm"); + final MockWebServer server = new MockWebServer(); + final ScheduledExecutorService sender = Executors.newSingleThreadScheduledExecutor(); + final ByteArrayOutputStream received = new ByteArrayOutputStream(); + final AtomicLong finalInputAt = new AtomicLong(); + Consumer update = socket -> socket.send("{\"type\":\"session.updated\"}"); + Consumer firstInput = socket -> {}; + Consumer finalInput = socket -> {}; + QwenLiveConnectionParam param; + + Scenario() throws Exception { + Files.write(input, new byte[] {1, 2, -1, -2}); + server.enqueue(new MockResponse().withWebSocketUpgrade(new WebSocketListener() { + @Override public void onMessage(WebSocket socket, String text) { + com.google.gson.JsonObject event = JsonUtils.parse(text); + String type = event.get("type").getAsString(); + if ("session.update".equals(type)) update.accept(socket); + if ("session.input_audio.append".equals(type)) { + boolean first = received.size() == 0; + byte[] bytes = Base64.getDecoder().decode(event.get("audio").getAsString()); + received.write(bytes, 0, bytes.length); + if (first) firstInput.accept(socket); + if (received.size() == 32004) { + finalInputAt.set(System.nanoTime()); + finalInput.accept(socket); + } + } + } + @Override public void onClosing(WebSocket socket, int code, String reason) { + socket.close(code, reason); + } + })); + server.start(); + param = QwenLiveConnectionParam.builder().model("qwen-live").apikey("test-key") + .url(server.url("/realtime").toString()).build(); + } + + void run(long idle, long timeout) throws Exception { + runSample(param, input, output, idle, timeout); + } + + @Override public void close() throws Exception { + sender.shutdownNow(); + check(sender.awaitTermination(2, TimeUnit.SECONDS), "test sender did not stop"); + server.shutdown(); + Files.deleteIfExists(input); + Files.deleteIfExists(output); + Files.delete(directory); + } + } + + private static void audio(WebSocket socket, String field, String data) { + socket.send("{\"type\":\"session.output_audio.delta\",\"" + field + "\":\"" + data + "\"}"); + } + + private static void savesExactAudioFragments() throws Exception { + try (Scenario scenario = new Scenario()) { + scenario.finalInput = socket -> { + audio(socket, "audio", "AAEC"); + audio(socket, "delta", "A/8="); + }; + scenario.run(50, 1000); + check(Arrays.equals(new byte[] {0, 1, 2, 3, -1}, Files.readAllBytes(scenario.output)), + "audio fragments must be saved exactly once"); + } + } + + private static void silenceWithoutAudioTimesOut() throws Exception { + try (Scenario scenario = new Scenario()) { + scenario.finalInput = socket -> { + audio(socket, "audio", ""); + socket.send("{\"type\":\"session.output_transcript.delta\",\"delta\":\"text only\"}"); + }; + try { + scenario.run(20, 200); + throw new AssertionError("empty audio/transcript must not count as completed output"); + } catch (TimeoutException expected) { + check(Files.size(scenario.output) == 0, "empty audio wrote unexpected bytes"); + } + } + } + + private static void newAudioRestartsIdleWindow() throws Exception { + try (Scenario scenario = new Scenario()) { + AtomicLong secondAudioAt = new AtomicLong(); + scenario.finalInput = socket -> { + audio(socket, "audio", "AQI="); + scenario.sender.schedule(() -> { + secondAudioAt.set(System.nanoTime()); + audio(socket, "audio", "AwQ="); + }, 180, TimeUnit.MILLISECONDS); + }; + scenario.run(250, 2000); + check(secondAudioAt.get() != 0, "sample closed before later audio"); + check(System.nanoTime() - secondAudioAt.get() >= TimeUnit.MILLISECONDS.toNanos(230), + "later audio must extend waiting"); + check(Arrays.equals(new byte[] {1, 2, 3, 4}, Files.readAllBytes(scenario.output)), + "later audio was lost"); + } + } + + private static void earlyAudioStillWaitsAfterInputEnds() throws Exception { + try (Scenario scenario = new Scenario()) { + scenario.firstInput = socket -> audio(socket, "audio", "AQI="); + scenario.run(180, 2000); + check(System.nanoTime() - scenario.finalInputAt.get() >= TimeUnit.MILLISECONDS.toNanos(250), + "audio during input must not cause immediate closure at EOF"); + } + } + + private static void errorsUnblockWaitingAndPreserveTheirCause() throws Exception { + for (String error : new String[] {"server", "transport", "session"}) { + try (Scenario scenario = new Scenario()) { + scenario.finalInput = socket -> { + if ("server".equals(error)) { + socket.send("{\"type\":\"error\",\"error\":{\"code\":\"test_error\",\"message\":\"request-failed\"}}"); + } else if ("transport".equals(error)) { + socket.close(1000, "unexpected close"); + } else { + socket.send("{\"type\":\"session.closed\",\"reason\":\"unexpected end\"}"); + } + }; + try { + scenario.run(50, 2000); + throw new AssertionError(error + " error cannot become successful output"); + } catch (IllegalStateException expected) { + check(expected.getCause() != null, "original failure must be preserved"); + if ("server".equals(error)) { + check(expected.getCause() instanceof com.alibaba.dashscope.exception.ApiException, + "server failure must retain ApiException"); + } else { + check(expected.getCause() instanceof IOException, "early closure must explain incomplete output"); + } + } + } + } + } + + private static void fileWriteFailureIsNotReportedAsSuccess() throws Exception { + ExecutorService worker = Executors.newSingleThreadExecutor(); + AtomicBoolean interruptAudio = new AtomicBoolean(); + try (Scenario scenario = new Scenario()) { + scenario.param.setCallbackExecutor(task -> worker.execute(() -> { + if (interruptAudio.getAndSet(false)) Thread.currentThread().interrupt(); + try { task.run(); } finally { Thread.interrupted(); } + })); + scenario.finalInput = socket -> { + interruptAudio.set(true); + audio(socket, "audio", Base64.getEncoder().encodeToString(new byte[10000])); + }; + try { + scenario.run(50, 2000); + throw new AssertionError("failed file write must not count as saved audio"); + } catch (IllegalStateException expected) { + check(expected.getCause() instanceof ClosedByInterruptException, + "real disk write failure must reach caller: " + expected); + } + } finally { + worker.shutdownNow(); + check(worker.awaitTermination(2, TimeUnit.SECONDS), "test worker did not stop"); + } + } + + private static void sessionUpdateMustBeConfirmed() throws Exception { + try (Scenario scenario = new Scenario()) { + AtomicBoolean confirmed = new AtomicBoolean(); + AtomicBoolean prematureInput = new AtomicBoolean(); + scenario.update = socket -> { + socket.send("{\"type\":\"session.created\",\"session\":{\"id\":\"test-session\"}}"); + scenario.sender.schedule(() -> { + confirmed.set(true); + socket.send("{\"type\":\"session.updated\"}"); + }, 120, TimeUnit.MILLISECONDS); + }; + scenario.firstInput = socket -> prematureInput.set(!confirmed.get()); + scenario.finalInput = socket -> audio(socket, "audio", "AQI="); + scenario.run(50, 1000); + check(confirmed.get() && !prematureInput.get(), "websocket/session.created is not configuration confirmation"); + } + } + + private static void malformedAudioIsNotReportedAsSuccess() throws Exception { + try (Scenario scenario = new Scenario()) { + scenario.finalInput = socket -> audio(socket, "audio", "not-valid-base64!"); + try { + scenario.run(50, 1000); + throw new AssertionError("malformed audio cannot become successful output"); + } catch (IllegalStateException expected) { + check(expected.getCause() instanceof IllegalArgumentException, "decoding error must reach caller"); + } + } + } + + private static Object staticCall(String name, Class[] types, Object... args) throws Exception { + Method method = Class.forName("QwenLiveQuickStart").getDeclaredMethod(name, types); + method.setAccessible(true); + try { return method.invoke(null, args); } + catch (InvocationTargetException wrapped) { + Throwable cause = wrapped.getCause(); + if (cause instanceof Exception) throw (Exception) cause; + throw new AssertionError(cause); + } + } + + private static void runSample(QwenLiveConnectionParam param, Path input, Path output, long idle, long timeout) + throws Exception { + staticCall("run", new Class[] {QwenLiveConnectionParam.class, String.class, Path.class, Path.class, long.class, long.class}, + param, "test-backend", input, output, idle, timeout); + } + + private static void doesNotOverwriteOutputAndRejectsBadWaitSettings() throws Exception { + Path directory = Files.createTempDirectory(Paths.get("target"), "quickstart-input-"); + Path input = directory.resolve("input.pcm"); + Path output = directory.resolve("existing.pcm"); + QwenLiveConnectionParam param = QwenLiveConnectionParam.builder().model("qwen-live").apikey("test-key").url("http://127.0.0.1:1/realtime").build(); + try { + Files.write(input, new byte[] {1, 2}); + Files.write(output, new byte[] {9, 8}); + try { runSample(param, input, output, 10, 500); throw new AssertionError("existing output must not be overwritten"); } + catch (java.nio.file.FileAlreadyExistsException expected) { } + check(Arrays.equals(new byte[] {9, 8}, Files.readAllBytes(output)), "existing output changed"); + try { runSample(param, input, output, 0, 500); throw new AssertionError("invalid wait settings accepted"); } + catch (IllegalArgumentException expected) { } + } finally { + Files.deleteIfExists(input); + Files.deleteIfExists(output); + Files.delete(directory); + } + } + + private static void fileToReplyLoopWaitsForDelayedAudio() throws Exception { + Path directory = Files.createTempDirectory(Paths.get("target"), "quickstart-loop-"); + Path input = directory.resolve("input.pcm"); + Path output = directory.resolve("reply.pcm"); + Files.write(input, new byte[] {1, 2, -1, -2}); + ByteArrayOutputStream receivedInput = new ByteArrayOutputStream(); + AtomicBoolean sentFinalAudio = new AtomicBoolean(); + AtomicBoolean prematureClose = new AtomicBoolean(); + AtomicBoolean correctDelegation = new AtomicBoolean(); + ScheduledExecutorService sender = Executors.newSingleThreadScheduledExecutor(); + MockWebServer server = new MockWebServer(); + server.enqueue(new MockResponse().withWebSocketUpgrade(new WebSocketListener() { + @Override public void onMessage(WebSocket socket, String text) { + com.google.gson.JsonObject event = JsonUtils.parse(text); + String type = event.get("type").getAsString(); + if ("session.update".equals(type)) { + correctDelegation.set("responses".equals(event.getAsJsonObject("session").getAsJsonObject("delegation").get("type").getAsString())); + socket.send("{\"type\":\"session.updated\"}"); + } else if ("session.input_audio.append".equals(type)) { + byte[] data = Base64.getDecoder().decode(event.get("audio").getAsString()); + receivedInput.write(data, 0, data.length); + if (receivedInput.size() == 32004) { + socket.send("{\"type\":\"session.output_audio.delta\",\"audio\":\"BQY=\"}"); + sender.schedule(() -> { + sentFinalAudio.set(true); + socket.send("{\"type\":\"session.output_audio.delta\",\"delta\":\"BwgJCg==\"}"); + }, 120, TimeUnit.MILLISECONDS); + } + } + } + @Override public void onClosing(WebSocket socket, int code, String reason) { + if (!sentFinalAudio.get()) prematureClose.set(true); + socket.close(code, reason); + } + })); + server.start(); + try { + QwenLiveConnectionParam param = QwenLiveConnectionParam.builder().model("qwen-live").apikey("test-key").url(server.url("/realtime").toString()).build(); + runSample(param, input, output, 250, 3000); + check(correctDelegation.get(), "minimal sample must use server responses delegation"); + check(!prematureClose.get() && sentFinalAudio.get(), "sample closed before delayed final audio"); + check(Arrays.equals(new byte[] {5, 6, 7, 8, 9, 10}, Files.readAllBytes(output)), "reply PCM did not contain all fragments"); + byte[] expectedInput = new byte[32004]; + System.arraycopy(new byte[] {1, 2, -1, -2}, 0, expectedInput, 0, 4); + check(Arrays.equals(expectedInput, receivedInput.toByteArray()), "input slice or trailing silence was encoded incorrectly"); + } finally { + sender.shutdownNow(); + check(sender.awaitTermination(2, TimeUnit.SECONDS), "test sender did not stop"); + server.shutdown(); + Files.deleteIfExists(input); + Files.deleteIfExists(output); + Files.delete(directory); + } + } + + private static void standaloneCliExitsAfterSuccessAndFailure() throws Exception { + Path directory = Files.createTempDirectory(Paths.get("target"), "quickstart-cli-"); + Path input = directory.resolve("input.pcm"); + Files.write(input, new byte[] {1, 2}); + MockWebServer server = new MockWebServer(); + server.start(); + try { + for (boolean fail : new boolean[] {false, true}) { + server.enqueue(new MockResponse().withWebSocketUpgrade(new WebSocketListener() { + @Override public void onMessage(WebSocket socket, String text) { + String type = JsonUtils.parse(text).get("type").getAsString(); + if ("session.update".equals(type)) { + socket.send(fail + ? "{\"type\":\"error\",\"error\":{\"code\":\"test_error\",\"message\":\"configuration failed\"}}" + : "{\"type\":\"session.updated\"}"); + } else if ("session.input_audio.append".equals(type)) { + socket.send("{\"type\":\"session.output_audio.delta\",\"audio\":\"AwQ=\"}"); + } + } + @Override public void onClosing(WebSocket socket, int code, String reason) { socket.close(code, reason); } + })); + Path output = directory.resolve(fail ? "failed.pcm" : "reply.pcm"); + ProcessBuilder command = new ProcessBuilder( + System.getProperty("java.home") + "/bin/java", "-cp", System.getProperty("java.class.path"), + "QwenLiveQuickStart", "test-workspace", "test-backend", input.toString(), output.toString(), "1", "3"); + command.environment().put("DASHSCOPE_API_KEY", "test-key"); + command.environment().put("DASHSCOPE_WEBSOCKET_OMNI_BASE_URL", server.url("/realtime").toString()); + command.redirectErrorStream(true); + Process process = command.start(); + try { + check(process.waitFor(7, TimeUnit.SECONDS), "CLI hung after closing its request"); + ByteArrayOutputStream log = new ByteArrayOutputStream(); + byte[] buffer = new byte[2048]; + int read; + while ((read = process.getInputStream().read(buffer)) != -1) log.write(buffer, 0, read); + String text = log.toString("UTF-8"); + check(process.exitValue() == (fail ? 1 : 0), "CLI returned incorrect exit status: " + text); + check(text.contains("Saved raw PCM:") != fail, "CLI reported failure as saved output: " + text); + if (!fail) check(Files.size(output) > 0, "successful CLI did not save audio"); + } finally { + if (process.isAlive()) process.destroyForcibly(); + process.waitFor(2, TimeUnit.SECONDS); + Files.deleteIfExists(output); + } + } + } finally { + server.shutdown(); + Files.deleteIfExists(input); + Files.delete(directory); + } + } + + private static void invalidCliArgumentsReturnFailure() throws Exception { + Process process = new ProcessBuilder(System.getProperty("java.home") + "/bin/java", "-cp", + System.getProperty("java.class.path"), "QwenLiveQuickStart").redirectErrorStream(true).start(); + try { + check(process.waitFor(5, TimeUnit.SECONDS), "usage path must exit immediately"); + check(process.exitValue() == 2, "invalid CLI argument count must exit 2, not indicate success"); + } finally { + if (process.isAlive()) process.destroyForcibly(); + process.waitFor(2, TimeUnit.SECONDS); + } + } + + private static void check(boolean condition, String message) { + if (!condition) throw new AssertionError(message); + } +} diff --git a/src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveConversation.java b/src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveConversation.java new file mode 100644 index 0000000..0eb2585 --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveConversation.java @@ -0,0 +1,840 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static org.junit.jupiter.api.Assertions.*; + +import com.google.gson.JsonObject; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.PrintStream; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import okhttp3.*; +import okhttp3.mockwebserver.*; +import okio.ByteString; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +public class TestQwenLiveConversation { + private MockWebServer server; + private QwenLiveConnectionParam param; + private RecordingCallback callback; + private QwenLiveConversation conversation; + private ExecutorService executor; + + @BeforeEach + void setUp() throws Exception { + server = new MockWebServer(); + server.start(); + param = + QwenLiveConnectionParam.builder() + .model("qwen-live") + .apikey("test-key") + .url(server.url("/realtime").toString()) + .build(); + callback = new RecordingCallback(); + executor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + void tearDown() throws Exception { + if (conversation != null) conversation.cancel(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + server.shutdown(); + } + + @Test + void unconnectedCloseIsSafe() throws Exception { + conversation = new QwenLiveConversation(param, callback); + conversation.close(); + conversation.close(); + conversation.cancel(); + assertFalse(conversation.awaitSessionClosed(0)); + assertThrows(IllegalStateException.class, conversation::connect); + assertThrows(IllegalStateException.class, conversation::checkStatus); + } + + @Test + void waitingBeforeConnectAndNegativeTimeout() throws Exception { + conversation = new QwenLiveConversation(param, callback); + assertFalse(conversation.awaitSessionClosed(0)); + assertThrows(IllegalArgumentException.class, () -> conversation.awaitSessionClosed(-1)); + } + + @Test + void ordinaryWebSocketCloseIsNotSessionConfirmation() throws Exception { + server.enqueue( + new MockResponse() + .withWebSocketUpgrade( + new WebSocketListener() { + @Override + public void onOpen(WebSocket socket, Response response) { + socket.close(1000, "bye"); + } + })); + conversation = new QwenLiveConversation(param, callback); + try { + conversation.connect(); + } catch (IllegalStateException expected) { + } + assertTrue(callback.transportClosed.await(5, TimeUnit.SECONDS)); + assertFalse(conversation.awaitSessionClosed(0)); + assertThrows(IllegalStateException.class, conversation::checkStatus); + } + + @Test + void rejectedHandshakeIsNotSuccessfulConnect() throws Exception { + server.enqueue(new MockResponse().setResponseCode(401).setBody("denied")); + conversation = new QwenLiveConversation(param, callback); + IllegalStateException failure = + assertThrows(IllegalStateException.class, conversation::connect); + assertNotNull(failure.getCause()); + assertTrue(callback.transportClosed.await(5, TimeUnit.SECONDS)); + assertFalse(conversation.awaitSessionClosed(0)); + assertThrows(IllegalStateException.class, conversation::connect); + } + + @Test + void handshakeHeadersKeepHostWorkspaceAndHeaderWorkspaceSeparate() throws Exception { + server.enqueue(new MockResponse().withWebSocketUpgrade(new WebSocketListener() {})); + QwenLiveConnectionParam withHeaders = + QwenLiveConnectionParam.builder() + .model("qwen-live") + .apikey("test-key") + .workspaceId("host-space") + .workspace("header-space") + .header("X-Test", 123) + .url(server.url("/realtime").toString()) + .build(); + conversation = new QwenLiveConversation(withHeaders, callback); + conversation.connect(); + RecordedRequest request = server.takeRequest(5, TimeUnit.SECONDS); + assertNotNull(request); + assertEquals("Bearer test-key", request.getHeader("Authorization")); + assertEquals("header-space", request.getHeader("X-DashScope-WorkSpace")); + assertEquals("123", request.getHeader("X-Test")); + assertEquals("/realtime?model=qwen-live", request.getPath()); + } + + @Test + void workspaceIdAloneDoesNotBecomeWorkspaceHeader() throws Exception { + FakeFactory factory = new FakeFactory(); + param.setWorkspaceId("host-space"); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + assertNull(factory.socket.request().header("X-DashScope-WorkSpace")); + assertEquals("Bearer test-key", factory.socket.request().header("Authorization")); + } + + @Test + void synchronousOnOpenBeforeFactoryReturnIsSupported() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + conversation.checkStatus(); + assertEquals(1, callback.openCalls.get()); + assertThrows(IllegalStateException.class, conversation::connect); + } + + @Test + void synchronousFailureCancelsReturnedSocket() throws Exception { + FakeFactory factory = new FakeFactory(); + factory.action = + (socket, listener) -> listener.onFailure(socket, new IOException("early"), null); + conversation = new QwenLiveConversation(param, callback, factory, 1); + assertThrows(IllegalStateException.class, conversation::connect); + assertTrue(factory.socket.cancelCalls.get() > 0); + assertFalse(conversation.awaitSessionClosed(0)); + } + + @Test + void cancellationInsideOnOpenCannotRestoreOpenState() throws Exception { + FakeFactory factory = new FakeFactory(); + QwenLiveCallback cancels = + new RecordingCallback() { + @Override + public void onOpen() { + conversation.cancel(); + } + }; + conversation = new QwenLiveConversation(param, cancels, factory, 1); + assertThrows(IllegalStateException.class, conversation::connect); + assertTrue(factory.socket.cancelCalls.get() > 0); + assertThrows(IllegalStateException.class, conversation::checkStatus); + } + + @Test + void cancellationDuringFactoryReturnCancelsLateSocket() throws Exception { + FakeFactory factory = new FakeFactory(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + factory.action = + (socket, listener) -> { + entered.countDown(); + boundedAwait(release); + listener.onOpen(socket, null); + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + Future result = + executor.submit(() -> assertThrows(IllegalStateException.class, conversation::connect)); + try { + assertTrue(entered.await(5, TimeUnit.SECONDS)); + conversation.cancel(); + } finally { + release.countDown(); + } + result.get(5, TimeUnit.SECONDS); + assertTrue(factory.socket.cancelCalls.get() > 0); + assertEquals(0, callback.openCalls.get()); + } + + @Test + void concurrentConnectIsRejectedWithoutSecondSocket() throws Exception { + FakeFactory factory = new FakeFactory(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + factory.action = + (socket, listener) -> { + entered.countDown(); + boundedAwait(release); + listener.onOpen(socket, null); + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + Future first = + executor.submit( + () -> { + try { + conversation.connect(); + } catch (Exception e) { + throw new AssertionError(e); + } + }); + try { + assertTrue(entered.await(5, TimeUnit.SECONDS)); + assertThrows(IllegalStateException.class, conversation::connect); + } finally { + release.countDown(); + } + first.get(5, TimeUnit.SECONDS); + assertEquals(1, factory.calls.get()); + } + + @Test + void timeoutCancelsSocketAndReleasesProtocolWait() throws Exception { + FakeFactory factory = new FakeFactory(); + factory.action = (socket, listener) -> {}; + conversation = new QwenLiveConversation(param, callback, factory, 1); + assertThrows(IllegalStateException.class, conversation::connect); + assertTrue(factory.socket.cancelCalls.get() > 0); + assertFalse(conversation.awaitSessionClosed(0)); + } + + @Test + void interruptDuringConnectRestoresFlagAndCancelsSocket() throws Exception { + FakeFactory factory = new FakeFactory(); + CountDownLatch entered = new CountDownLatch(1); + AtomicReference worker = new AtomicReference<>(); + factory.action = (socket, listener) -> entered.countDown(); + conversation = new QwenLiveConversation(param, callback, factory, 10); + Future result = + executor.submit( + () -> { + worker.set(Thread.currentThread()); + try { + conversation.connect(); + return false; + } catch (InterruptedException expected) { + return Thread.currentThread().isInterrupted(); + } + }); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + worker.get().interrupt(); + assertTrue(result.get(5, TimeUnit.SECONDS)); + assertTrue(factory.socket.cancelCalls.get() > 0); + } + + @Test + void closeDuringConnectReleasesWaiters() throws Exception { + FakeFactory factory = new FakeFactory(); + CountDownLatch entered = new CountDownLatch(1); + factory.action = (socket, listener) -> entered.countDown(); + conversation = new QwenLiveConversation(param, callback, factory, 10); + Future result = + executor.submit(() -> assertThrows(IllegalStateException.class, conversation::connect)); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + conversation.close(); + result.get(5, TimeUnit.SECONDS); + assertTrue(factory.socket.cancelCalls.get() > 0); + } + + @Test + void closeThenCancelIsIdempotentButStillForcesSocketRelease() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + conversation.close(); + conversation.close(); + assertEquals(1, factory.socket.closeCalls.get()); + conversation.cancel(); + conversation.cancel(); + assertEquals(1, factory.socket.cancelCalls.get()); + } + + @ParameterizedTest + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + void gracefulCloseKeepsWaitingForDelayedSessionConfirmation( + boolean alreadyWaiting, boolean repeatClose) throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + Future result = alreadyWaiting ? startSessionClosedWait() : null; + + conversation.close(); + if (repeatClose) conversation.close(); + assertThrows(IllegalStateException.class, () -> conversation.appendAudio("AAA=")); + if (result == null) result = startSessionClosedWait(); + Future waiting = result; + assertThrows(TimeoutException.class, () -> waiting.get(100, TimeUnit.MILLISECONDS)); + + factory.listener.onMessage( + factory.socket, "{\"type\":\"session.closed\",\"reason\":\"close_requested\"}"); + assertTrue(waiting.get(1, TimeUnit.SECONDS)); + assertEquals("close_requested", conversation.getCloseReason()); + assertTrue(conversation.awaitSessionClosed(0)); + } + + @Test + void gracefulCloseWaitsUntilTimeoutWithoutConfirmation() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + conversation.close(); + assertFalse(conversation.awaitSessionClosed(0)); + + long started = System.nanoTime(); + assertFalse(conversation.awaitSessionClosed(1)); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started); + assertTrue(elapsedMillis >= 900, "Protocol wait returned early: " + elapsedMillis + " ms"); + } + + @ParameterizedTest + @ValueSource(strings = {"cancel", "closing", "closed", "failure"}) + void terminalTransportAfterGracefulCloseReleasesProtocolWait(String terminal) throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + conversation.close(); + Future result = startSessionClosedWait(); + assertThrows(TimeoutException.class, () -> result.get(100, TimeUnit.MILLISECONDS)); + + switch (terminal) { + case "cancel": + conversation.cancel(); + break; + case "closing": + factory.listener.onClosing(factory.socket, 1000, "bye"); + break; + case "closed": + factory.listener.onClosed(factory.socket, 1000, "bye"); + break; + case "failure": + factory.listener.onFailure(factory.socket, new IOException("connection lost"), null); + break; + default: + fail("Unknown terminal event: " + terminal); + } + assertFalse(result.get(1, TimeUnit.SECONDS)); + assertFalse(executor.submit(() -> conversation.awaitSessionClosed(30)).get(1, TimeUnit.SECONDS)); + } + + private Future startSessionClosedWait() throws Exception { + CountDownLatch started = new CountDownLatch(1); + Future result = + executor.submit( + () -> { + started.countDown(); + return conversation.awaitSessionClosed(30); + }); + assertTrue(started.await(5, TimeUnit.SECONDS)); + return result; + } + + @Test + void rejectedCloseFallsBackToCancel() throws Exception { + FakeFactory factory = new FakeFactory(); + factory.socket.closeResult = false; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + conversation.close(); + assertTrue(factory.socket.cancelCalls.get() > 0); + assertFalse(executor.submit(() -> conversation.awaitSessionClosed(30)).get(1, TimeUnit.SECONDS)); + } + + @Test + void invalidCloseArgumentsDoNotCloseConversation() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + for (int code : new int[] {999, 1005, 1006, 1015, 2000, 5000}) { + assertThrows(IllegalArgumentException.class, () -> conversation.close(code, "bad")); + } + StringBuilder reason = new StringBuilder(); + for (int i = 0; i < 42; i++) reason.append('中'); + assertThrows(IllegalArgumentException.class, () -> conversation.close(1000, reason.toString())); + conversation.checkStatus(); + assertEquals(0, factory.socket.closeCalls.get()); + } + + @Test + void factoryExceptionIsTerminalAndReleasesWaiters() throws Exception { + conversation = + new QwenLiveConversation( + param, + callback, + (request, listener) -> { + throw new IllegalStateException("factory failed"); + }, + 1); + assertThrows(IllegalStateException.class, conversation::connect); + assertFalse(conversation.awaitSessionClosed(0)); + assertThrows(IllegalStateException.class, conversation::connect); + } + + @Test + void interruptDuringProtocolWaitDoesNotCancelConnection() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + Future result = + executor.submit( + () -> { + Thread.currentThread().interrupt(); + try { + conversation.awaitSessionClosed(1); + return false; + } catch (InterruptedException expected) { + return Thread.currentThread().isInterrupted(); + } + }); + assertTrue(result.get(5, TimeUnit.SECONDS)); + conversation.checkStatus(); + assertEquals(0, factory.socket.cancelCalls.get()); + } + + @Test + void throwingCloseAndErrorCallbacksDoNotBlockStateRelease() throws Exception { + FakeFactory factory = new FakeFactory(); + QwenLiveCallback throwing = + new RecordingCallback() { + @Override + public void onClose(int code, String reason) { + throw new IllegalStateException("callback-close"); + } + + @Override + public void onError(Throwable failure) { + throw new IllegalStateException("callback-error"); + } + }; + conversation = new QwenLiveConversation(param, throwing, factory, 1); + conversation.connect(); + assertDoesNotThrow(() -> factory.listener.onClosed(factory.socket, 1000, "bye")); + assertDoesNotThrow( + () -> factory.listener.onFailure(factory.socket, new IOException("failure"), null)); + assertFalse(conversation.awaitSessionClosed(0)); + assertThrows(IllegalStateException.class, conversation::checkStatus); + } + + @Test + void defaultErrorCallbackLogsUnhandledFailure() throws Exception { + QwenLiveCallback minimal = + new QwenLiveCallback() { + @Override + public void onEvent(JsonObject event) {} + + @Override + public void onClose(int code, String reason) {} + }; + ByteArrayOutputStream output = new ByteArrayOutputStream(); + PrintStream previous = System.err; + try (PrintStream captured = new PrintStream(output, true, "UTF-8")) { + System.setErr(captured); + minimal.onError(new IOException("unhandled-live-transport-failure")); + } finally { + System.setErr(previous); + } + assertTrue(output.toString("UTF-8").contains("unhandled-live-transport-failure")); + } + + @Test + void asyncCallbacksDoNotBlockTransportAndKeepEventOrder() throws Exception { + ExecutorService callbackExecutor = Executors.newFixedThreadPool(2); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + BlockingQueue calls = new LinkedBlockingQueue<>(); + try { + enableAsyncCallbacks(callbackExecutor, 8); + FakeFactory factory = new FakeFactory(); + callback = + new RecordingCallback() { + @Override + public void onOpen() { + calls.add("open"); + } + + @Override + public void onEvent(JsonObject event) { + String type = event.get("type").getAsString(); + if ("slow.event".equals(type)) { + entered.countDown(); + boundedAwait(release); + } + calls.add(type); + } + + @Override + public void onClose(int code, String reason) { + calls.add("transport.closed"); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + assertEquals("open", calls.poll(1, TimeUnit.SECONDS)); + Future received = + executor.submit( + () -> factory.listener.onMessage(factory.socket, "{\"type\":\"slow.event\"}")); + assertTrue(entered.await(1, TimeUnit.SECONDS)); + assertDoesNotThrow(() -> received.get(1, TimeUnit.SECONDS)); + + factory.listener.onMessage( + factory.socket, "{\"type\":\"session.created\",\"session\":{\"id\":\"async-session\"}}"); + assertEquals("async-session", conversation.getSessionId()); + conversation.close(); + factory.listener.onMessage( + factory.socket, "{\"type\":\"session.closed\",\"reason\":\"close_requested\"}"); + assertTrue(conversation.awaitSessionClosed(0)); + factory.listener.onClosed(factory.socket, 1000, "bye"); + assertNull(calls.poll(100, TimeUnit.MILLISECONDS)); + + release.countDown(); + assertEquals("slow.event", calls.poll(1, TimeUnit.SECONDS)); + assertEquals("session.created", calls.poll(1, TimeUnit.SECONDS)); + assertEquals("session.closed", calls.poll(1, TimeUnit.SECONDS)); + assertEquals("transport.closed", calls.poll(1, TimeUnit.SECONDS)); + assertEquals("still-owned", callbackExecutor.submit(() -> "still-owned").get(1, TimeUnit.SECONDS)); + } finally { + release.countDown(); + callbackExecutor.shutdownNow(); + assertTrue(callbackExecutor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void asyncCallbackFailureIsReportedBeforeNextEvent() throws Exception { + ExecutorService callbackExecutor = Executors.newFixedThreadPool(2); + BlockingQueue calls = new LinkedBlockingQueue<>(); + try { + enableAsyncCallbacks(callbackExecutor, 8); + FakeFactory factory = new FakeFactory(); + callback = + new RecordingCallback() { + @Override + public void onEvent(JsonObject event) { + String type = event.get("type").getAsString(); + if ("bad.event".equals(type)) throw new IllegalStateException("callback-failed"); + calls.add(type); + } + + @Override + public void onError(Throwable failure) { + calls.add(failure.getMessage()); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + factory.listener.onMessage(factory.socket, "{\"type\":\"bad.event\"}"); + factory.listener.onMessage(factory.socket, "{\"type\":\"next.event\"}"); + assertEquals("callback-failed", calls.poll(1, TimeUnit.SECONDS)); + assertEquals("next.event", calls.poll(1, TimeUnit.SECONDS)); + conversation.checkStatus(); + } finally { + callbackExecutor.shutdownNow(); + assertTrue(callbackExecutor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void assertionErrorsInCallbacksDoNotStrandAsyncQueue(boolean errorHandlerThrows) throws Exception { + ExecutorService callbackExecutor = Executors.newSingleThreadExecutor(); + BlockingQueue calls = new LinkedBlockingQueue<>(); + try { + enableAsyncCallbacks(callbackExecutor, 8); + FakeFactory factory = new FakeFactory(); + callback = + new RecordingCallback() { + @Override + public void onEvent(JsonObject event) { + String type = event.get("type").getAsString(); + if ("bad.event".equals(type)) { + if (errorHandlerThrows) throw new IllegalStateException("callback-failed"); + throw new AssertionError("callback-failed"); + } + calls.add(type); + } + + @Override + public void onError(Throwable failure) { + calls.add(failure.getMessage()); + if (errorHandlerThrows) throw new AssertionError("error-handler-failed"); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + factory.listener.onMessage(factory.socket, "{\"type\":\"bad.event\"}"); + factory.listener.onMessage(factory.socket, "{\"type\":\"next.event\"}"); + assertEquals("callback-failed", calls.poll(1, TimeUnit.SECONDS)); + assertEquals("next.event", calls.poll(1, TimeUnit.SECONDS)); + conversation.checkStatus(); + } finally { + callbackExecutor.shutdownNow(); + assertTrue(callbackExecutor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void rejectedCallbackExecutorCancelsWithoutCallingBusinessCodeOnTransportThread() throws Exception { + enableAsyncCallbacks(task -> { throw new RejectedExecutionException("executor-unavailable"); }, 8); + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + assertThrows(IllegalStateException.class, conversation::connect); + assertThrows(IllegalStateException.class, conversation::checkStatus); + assertFalse(conversation.awaitSessionClosed(0)); + assertEquals(0, callback.openCalls.get()); + assertTrue(callback.errors.isEmpty()); + } + + @Test + void fullAsyncCallbackQueueCancelsAndReportsErrorOnCallbackThread() throws Exception { + ExecutorService callbackExecutor = Executors.newSingleThreadExecutor(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference callbackThread = new AtomicReference<>(); + AtomicReference errorThread = new AtomicReference<>(); + BlockingQueue deliveredEvents = new LinkedBlockingQueue<>(); + try { + enableAsyncCallbacks(callbackExecutor, 1); + FakeFactory factory = new FakeFactory(); + callback = + new RecordingCallback() { + @Override + public void onOpen() { + callbackThread.set(Thread.currentThread()); + entered.countDown(); + boundedAwait(release); + } + + @Override + public void onEvent(JsonObject event) { + deliveredEvents.add(event.get("type").getAsString()); + } + + @Override + public void onError(Throwable failure) { + errorThread.set(Thread.currentThread()); + errors.add(failure); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + assertTrue(entered.await(1, TimeUnit.SECONDS)); + factory.listener.onMessage(factory.socket, "{\"type\":\"queued.event\"}"); + factory.listener.onMessage(factory.socket, "{\"type\":\"overflow.event\"}"); + assertThrows(IllegalStateException.class, conversation::checkStatus); + assertFalse(conversation.awaitSessionClosed(0)); + release.countDown(); + assertInstanceOf(RejectedExecutionException.class, callback.errors.poll(1, TimeUnit.SECONDS)); + assertEquals(callbackThread.get(), errorThread.get()); + assertTrue(deliveredEvents.isEmpty()); + } finally { + release.countDown(); + callbackExecutor.shutdownNow(); + assertTrue(callbackExecutor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void queueOverflowReschedulesErrorWhenWorkerExitsDuringCancellation() throws Exception { + ExecutorService callbackExecutor = Executors.newSingleThreadExecutor(); + CountDownLatch callbackEntered = new CountDownLatch(1); + CountDownLatch callbackRelease = new CountDownLatch(1); + CountDownLatch cancelEntered = new CountDownLatch(1); + CountDownLatch cancelRelease = new CountDownLatch(1); + AtomicReference listener = new AtomicReference<>(); + try { + enableAsyncCallbacks(callbackExecutor, 1); + FakeSocket socket = + new FakeSocket() { + @Override + public void cancel() { + super.cancel(); + cancelEntered.countDown(); + boundedAwait(cancelRelease); + } + }; + WebSocket.Factory factory = + (request, connectionListener) -> { + listener.set(connectionListener); + connectionListener.onOpen(socket, null); + return socket; + }; + callback = + new RecordingCallback() { + @Override + public void onOpen() { + callbackEntered.countDown(); + boundedAwait(callbackRelease); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + assertTrue(callbackEntered.await(1, TimeUnit.SECONDS)); + listener.get().onMessage(socket, "{\"type\":\"queued.event\"}"); + Future overflow = + executor.submit(() -> listener.get().onMessage(socket, "{\"type\":\"overflow.event\"}")); + assertTrue(cancelEntered.await(1, TimeUnit.SECONDS)); + assertThrows(IllegalStateException.class, conversation::checkStatus); + callbackRelease.countDown(); + callbackExecutor.submit(() -> {}).get(1, TimeUnit.SECONDS); + assertTrue(callback.errors.isEmpty()); + cancelRelease.countDown(); + overflow.get(1, TimeUnit.SECONDS); + assertInstanceOf(RejectedExecutionException.class, callback.errors.poll(1, TimeUnit.SECONDS)); + assertTrue(callback.events.isEmpty()); + assertFalse(conversation.awaitSessionClosed(0)); + } finally { + callbackRelease.countDown(); + cancelRelease.countDown(); + callbackExecutor.shutdownNow(); + assertTrue(callbackExecutor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @ParameterizedTest + @ValueSource(ints = {0, -1}) + void invalidCallbackQueueCapacityIsRejectedBeforeConnecting(int capacity) { + param.setCallbackQueueCapacity(capacity); + assertThrows( + IllegalArgumentException.class, + () -> new QwenLiveConversation(param, callback, new FakeFactory(), 1)); + } + + private void enableAsyncCallbacks(Executor callbackExecutor, int capacity) { + param.setCallbackExecutor(callbackExecutor); + param.setCallbackQueueCapacity(capacity); + } + + static void boundedAwait(CountDownLatch latch) { + try { + assertTrue(latch.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + + static class RecordingCallback extends QwenLiveCallback { + final CountDownLatch transportClosed = new CountDownLatch(1); + final AtomicInteger openCalls = new AtomicInteger(); + final BlockingQueue events = new LinkedBlockingQueue<>(); + final BlockingQueue errors = new LinkedBlockingQueue<>(); + + @Override + public void onOpen() { + openCalls.incrementAndGet(); + } + + @Override + public void onEvent(JsonObject message) { + events.add(message); + } + + @Override + public void onClose(int code, String reason) { + transportClosed.countDown(); + } + + @Override + public void onError(Throwable failure) { + errors.add(failure); + transportClosed.countDown(); + } + } + + interface SocketAction { + void accept(FakeSocket socket, WebSocketListener listener); + } + + static class FakeFactory implements WebSocket.Factory { + final FakeSocket socket = new FakeSocket(); + final AtomicInteger calls = new AtomicInteger(); + WebSocketListener listener; + SocketAction action = (socket, listener) -> listener.onOpen(socket, null); + + @Override + public WebSocket newWebSocket(Request request, WebSocketListener listener) { + calls.incrementAndGet(); + this.listener = listener; + socket.request = request; + action.accept(socket, listener); + return socket; + } + } + + static class FakeSocket implements WebSocket { + Request request; + boolean sendResult = true; + boolean closeResult = true; + final AtomicInteger cancelCalls = new AtomicInteger(); + final AtomicInteger closeCalls = new AtomicInteger(); + final BlockingQueue sent = new LinkedBlockingQueue<>(); + + @Override + public Request request() { + return request; + } + + @Override + public long queueSize() { + return 0; + } + + @Override + public boolean send(String text) { + if (sendResult) sent.add(text); + return sendResult; + } + + @Override + public boolean send(ByteString bytes) { + return sendResult; + } + + @Override + public boolean close(int code, String reason) { + closeCalls.incrementAndGet(); + return closeResult; + } + + @Override + public void cancel() { + cancelCalls.incrementAndGet(); + } + } +} diff --git a/src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveProtocol.java b/src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveProtocol.java new file mode 100644 index 0000000..436fdd0 --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/audio/live/TestQwenLiveProtocol.java @@ -0,0 +1,553 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. +package com.alibaba.dashscope.audio.live; + +import static com.alibaba.dashscope.audio.live.QwenLiveConstants.*; +import static org.junit.jupiter.api.Assertions.*; + +import com.alibaba.dashscope.audio.live.TestQwenLiveConversation.FakeFactory; +import com.alibaba.dashscope.audio.live.TestQwenLiveConversation.RecordingCallback; +import com.alibaba.dashscope.exception.ApiException; +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import java.util.stream.Stream; +import okhttp3.*; +import okhttp3.mockwebserver.*; +import org.junit.jupiter.api.*; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +public class TestQwenLiveProtocol { + private MockWebServer server; + private QwenLiveConnectionParam param; + private RecordingCallback callback; + private QwenLiveConversation conversation; + private final BlockingQueue sent = new LinkedBlockingQueue<>(); + private final BlockingQueue serverSocket = new LinkedBlockingQueue<>(); + + @BeforeEach + void setUp() throws Exception { + server = new MockWebServer(); + server.start(); + param = + QwenLiveConnectionParam.builder() + .model("qwen-live") + .apikey("test-key") + .url(server.url("/realtime").toString()) + .build(); + callback = new RecordingCallback(); + server.enqueue( + new MockResponse() + .withWebSocketUpgrade( + new WebSocketListener() { + @Override + public void onOpen(WebSocket socket, Response response) { + serverSocket.add(socket); + } + + @Override + public void onMessage(WebSocket socket, String text) { + sent.add(JsonUtils.parse(text)); + } + @Override + public void onClosing(WebSocket socket, int code, String reason) { + socket.close(code, reason); + } + })); + } + + @AfterEach + void tearDown() throws Exception { + if (conversation != null) conversation.cancel(); + server.shutdown(); + } + + private void connect() throws Exception { + conversation = new QwenLiveConversation(param, callback); + conversation.connect(); + } + + private String verify(Supplier send, String expected) throws Exception { + String id = send.get(); + JsonObject actual = sent.poll(5, TimeUnit.SECONDS); + assertNotNull(actual); + assertEquals(id, actual.remove(FIELD_EVENT_ID).getAsString()); + assertTrue(id.matches("event_[0-9a-f]{32}")); + assertEquals(JsonUtils.parse(expected), actual); + return id; + } + + @Test + void everyClientCommandHasExactPayloadAndUniqueEventId() throws Exception { + connect(); + Set ids = new HashSet<>(); + QwenLiveSessionConfig config = + QwenLiveSessionConfig.builder() + .model("front") + .delegation(QwenLiveDelegation.builder().type(QwenLiveDelegationType.CLIENT).build()) + .build(); + ids.add( + verify( + () -> conversation.updateSession(config), + "{\"type\":\"session.update\",\"session\":{\"model\":\"front\",\"delegation\":{\"type\":\"client\"}}}")); + ids.add( + verify( + () -> conversation.appendAudio("AAAAAA=="), + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAAAAA==\"}")); + ids.add(verify(() -> conversation.muteInputAudio(), "{\"type\":\"session.input_audio.mute\"}")); + ids.add( + verify(() -> conversation.unmuteInputAudio(), "{\"type\":\"session.input_audio.unmute\"}")); + ids.add( + verify( + () -> conversation.appendInstruction("call_xxx", "提示词"), + "{\"type\":\"session.instruction.append\",\"delegation_id\":\"call_xxx\",\"content\":\"提示词\"}")); + ids.add( + verify( + () -> conversation.appendThinking("call_xxx", "思考"), + "{\"type\":\"session.thinking.append\",\"delegation_id\":\"call_xxx\",\"content\":\"思考\"}")); + ids.add( + verify( + () -> conversation.appendCommentary("call_xxx", "结果"), + "{\"type\":\"session.commentary.append\",\"delegation_id\":\"call_xxx\",\"content\":\"结果\"}")); + JsonObject item = + JsonUtils.parse( + "{\"type\":\"message\",\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"你好\"}]}"); + JsonObject original = item.deepCopy(); + ids.add( + verify( + () -> conversation.createResponseItem(item), + "{\"type\":\"response.item.create\",\"item\":" + item + "}")); + assertEquals(original, item); + ids.add(verify(() -> conversation.createResponse(), "{\"type\":\"response.create\"}")); + ids.add( + verify( + () -> + conversation.sendRaw( + "{\"type\":\"future.event\",\"event_id\":\"old\",\"data\":{\"value\":\"中文\",\"nullable\":null}}"), + "{\"type\":\"future.event\",\"data\":{\"value\":\"中文\",\"nullable\":null}}")); + assertEquals(10, ids.size()); + } + + @Test + void emptyContentAndRawAudioAreAllowed() throws Exception { + connect(); + verify( + () -> conversation.appendCommentary("call", ""), + "{\"type\":\"session.commentary.append\",\"delegation_id\":\"call\",\"content\":\"\"}"); + verify( + () -> + conversation.sendRaw( + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAAAAA==\"}"), + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAAAAA==\"}"); + } + + @Test + void appendAudioBytesFullArrayHasExactPayload() throws Exception { + connect(); + byte[] audio = {0, 1, 2, (byte) 0xfe, (byte) 0xff}; + verify( + () -> conversation.appendAudioBytes(audio), + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAEC/v8=\"}"); + assertArrayEquals(new byte[] {0, 1, 2, (byte) 0xfe, (byte) 0xff}, audio); + } + + @Test + void appendAudioBytesSliceEncodesOnlySelectedBytes() throws Exception { + connect(); + byte[] audio = {85, 0, 1, 2, (byte) 0xfe, (byte) 0xff, 102}; + verify( + () -> conversation.appendAudioBytes(audio, 1, 5), + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAEC/v8=\"}"); + verify( + () -> conversation.appendAudioBytes(audio, 4, 3), + "{\"type\":\"session.input_audio.append\",\"audio\":\"/v9m\"}"); + assertArrayEquals(new byte[] {85, 0, 1, 2, (byte) 0xfe, (byte) 0xff, 102}, audio); + } + + @Test + void appendAudioBytesPayloadSurvivesArrayReuse() throws Exception { + connect(); + byte[] audio = {0, 1, 2, (byte) 0xfe, (byte) 0xff}; + verify( + () -> { + String id = conversation.appendAudioBytes(audio); + Arrays.fill(audio, (byte) 0); + return id; + }, + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAEC/v8=\"}"); + byte[] slice = {85, 0, 1, 2, (byte) 0xfe, (byte) 0xff, 102}; + verify( + () -> { + String id = conversation.appendAudioBytes(slice, 1, 5); + Arrays.fill(slice, (byte) 0); + return id; + }, + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAEC/v8=\"}"); + } + + @Test + void appendAudioBytesRejectsInvalidInputBeforeSend() throws Exception { + connect(); + assertThrows(IllegalArgumentException.class, () -> conversation.appendAudioBytes(null)); + assertThrows(IllegalArgumentException.class, () -> conversation.appendAudioBytes(null, 0, 0)); + byte[] audio = {0, 1, 2}; + for (int[] range : + new int[][] { + {-1, 0}, + {0, -1}, + {-1, -1}, + {4, 0}, + {3, 1}, + {2, 2}, + {0, 4}, + {Integer.MAX_VALUE, 0}, + {Integer.MAX_VALUE, 1}, + {1, Integer.MAX_VALUE}, + {0, Integer.MAX_VALUE}, + {Integer.MAX_VALUE, Integer.MAX_VALUE} + }) { + assertThrows( + IllegalArgumentException.class, + () -> conversation.appendAudioBytes(audio, range[0], range[1]), + "range: " + Arrays.toString(range)); + } + assertThrows(IllegalArgumentException.class, () -> conversation.appendAudioBytes(new byte[0], 1, 0)); + verify( + () -> conversation.appendAudio("AAA="), + "{\"type\":\"session.input_audio.append\",\"audio\":\"AAA=\"}"); + assertNull(sent.poll(200, TimeUnit.MILLISECONDS)); + } + + @Test + void appendAudioBytesDisconnectedNeverReturnsSuccessId() throws Exception { + conversation = new QwenLiveConversation(param, callback); + byte[] audio = {0, 1, 2}; + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio)); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio, 1, 2)); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(new byte[0])); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio, 3, 0)); + conversation.connect(); + conversation.close(); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio)); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio, 1, 2)); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(new byte[0])); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio, 3, 0)); + assertNull(sent.poll(200, TimeUnit.MILLISECONDS)); + } + + @Test + void appendAudioBytesFailedEnqueueNeverReturnsSuccessId() throws Exception { + FakeFactory factory = new FakeFactory(); + factory.socket.sendResult = false; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + byte[] audio = {0, 1, 2}; + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio)); + assertThrows(IllegalStateException.class, () -> conversation.appendAudioBytes(audio, 1, 2)); + assertTrue(factory.socket.sent.isEmpty()); + } + + @Test + void appendAudioBytesAllowsEmptyArraysAndValidEmptySlices() throws Exception { + connect(); + String expected = "{\"type\":\"session.input_audio.append\",\"audio\":\"\"}"; + verify(() -> conversation.appendAudioBytes(new byte[0]), expected); + verify(() -> conversation.appendAudioBytes(new byte[0], 0, 0), expected); + byte[] audio = {0, 1, 2}; + verify(() -> conversation.appendAudioBytes(audio, 0, 0), expected); + verify(() -> conversation.appendAudioBytes(audio, 1, 0), expected); + verify(() -> conversation.appendAudioBytes(audio, audio.length, 0), expected); + verify(() -> conversation.appendAudio(""), expected); + } + + @Test + void invalidPayloadIsRejectedBeforeSend() throws Exception { + connect(); + assertThrows(IllegalArgumentException.class, () -> conversation.updateSession(null)); + assertThrows(IllegalArgumentException.class, () -> conversation.appendAudio(null)); + assertThrows(IllegalArgumentException.class, () -> conversation.createResponseItem(null)); + assertThrows( + IllegalArgumentException.class, () -> conversation.appendInstruction("call", null)); + assertThrows(IllegalArgumentException.class, () -> conversation.appendThinking("call", null)); + assertThrows(IllegalArgumentException.class, () -> conversation.appendCommentary("call", null)); + for (String id : new String[] {null, "", " "}) { + assertThrows(IllegalArgumentException.class, () -> conversation.appendInstruction(id, "x")); + assertThrows(IllegalArgumentException.class, () -> conversation.appendThinking(id, "x")); + assertThrows(IllegalArgumentException.class, () -> conversation.appendCommentary(id, "x")); + } + for (String raw : + new String[] { + null, + "", + "{", + "[]", + "null", + "{}", + "{\"type\":1}", + "{\"type\":null}", + "{\"type\":\" \"}", + "{type:'legacy'}", + "{\"type\":\"x\"} {}" + }) { + assertThrows(IllegalArgumentException.class, () -> conversation.sendRaw(raw), "raw: " + raw); + } + assertTrue(sent.isEmpty()); + } + + @Test + void disconnectedSendNeverReturnsSuccessId() throws Exception { + conversation = new QwenLiveConversation(param, callback); + assertThrows(IllegalStateException.class, () -> conversation.appendAudio("AAA=")); + assertThrows(IllegalStateException.class, () -> conversation.sendRaw("{\"type\":\"x\"}")); + conversation.connect(); + conversation.close(); + assertThrows(IllegalStateException.class, conversation::createResponse); + } + + @Test + void failedSocketEnqueueNeverReturnsSuccessId() throws Exception { + FakeFactory factory = new FakeFactory(); + factory.socket.sendResult = false; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + assertThrows(IllegalStateException.class, () -> conversation.appendAudio("AAA=")); + assertThrows(IllegalStateException.class, conversation::createResponse); + } + + @Test + void onOpenCanSendBeforeFactoryReturns() throws Exception { + FakeFactory factory = new FakeFactory(); + AtomicReference id = new AtomicReference<>(); + callback = + new RecordingCallback() { + @Override + public void onOpen() { + id.set(conversation.appendAudio("AAA=")); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + assertNotNull(id.get()); + assertEquals( + id.get(), + JsonUtils.parse(factory.socket.sent.poll(5, TimeUnit.SECONDS)) + .get("event_id") + .getAsString()); + } + + static Stream serverEvents() { + return Stream.of( + EVENT_ERROR, + EVENT_SESSION_CREATED, + EVENT_SESSION_UPDATED, + EVENT_SESSION_INPUT_AUDIO_MUTED, + EVENT_SESSION_INPUT_AUDIO_UNMUTED, + EVENT_SESSION_OUTPUT_AUDIO_DELTA, + EVENT_SESSION_INPUT_TRANSCRIPT_DELTA, + EVENT_SESSION_OUTPUT_TRANSCRIPT_DELTA, + EVENT_RESPONSE_EVENT, + EVENT_SESSION_USAGE_UPDATED, + EVENT_SESSION_CLOSED, + EVENT_SESSION_DELEGATION_CREATED, + EVENT_SESSION_INSTRUCTIONS_APPENDED, + EVENT_SESSION_THINKING_APPENDED, + EVENT_SESSION_COMMENTARY_APPENDED, + "future.server.event"); + } + + @ParameterizedTest + @MethodSource("serverEvents") + void allServerEventsArePassedThroughWithoutLosingFields(String type) throws Exception { + connect(); + JsonObject event = + JsonUtils.parse( + "{\"event_id\":\"server-id\",\"client_event_id\":\"client-id\"," + + "\"delegation_id\":\"call\",\"offset_ms\":3600,\"start_ms\":1000,\"end_ms\":1400," + + "\"arguments\":\"{\\\"city\\\":\\\"杭州\\\"}\",\"delta\":\"中文\"," + + "\"session\":{\"id\":\"sess-1\",\"status\":\"active\"},\"reason\":\"expired\"," + + "\"error\":{\"type\":\"server_error\",\"code\":\"test\"}," + + "\"usage\":{\"seconds\":32.5},\"context_window\":{\"usage_ratio\":0.12}," + + "\"event\":{\"type\":\"response.output_text.delta\",\"delta\":\"结果\",\"logprobs\":[]}}"); + event.addProperty("type", type); + WebSocket socket = serverSocket.poll(5, TimeUnit.SECONDS); + assertNotNull(socket); + assertTrue(socket.send(event.toString())); + JsonObject actual = callback.events.poll(5, TimeUnit.SECONDS); + assertEquals(event, actual); + if (EVENT_SESSION_CREATED.equals(type)) assertEquals("sess-1", conversation.getSessionId()); + if (EVENT_SESSION_CLOSED.equals(type)) { + assertEquals("expired", conversation.getCloseReason()); + assertTrue(conversation.awaitSessionClosed(0)); + assertThrows(IllegalStateException.class, conversation::createResponse); + } else { + conversation.checkStatus(); + assertFalse(conversation.awaitSessionClosed(0)); + } + } + + @Test + void stateIsUpdatedBeforeCallbackEvenWhenCallbackThrows() throws Exception { + FakeFactory factory = new FakeFactory(); + callback = + new RecordingCallback() { + @Override + public void onEvent(JsonObject event) { + if (EVENT_SESSION_CREATED.equals(event.get("type").getAsString())) { + assertEquals("sess-2", conversation.getSessionId()); + } else { + assertEquals("connection_lost", conversation.getCloseReason()); + assertThrows(IllegalStateException.class, conversation::checkStatus); + } + throw new IllegalStateException("event-callback"); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + factory.listener.onMessage( + factory.socket, "{\"type\":\"session.created\",\"session\":{\"id\":\"sess-2\"}}"); + assertNotNull(callback.errors.poll(5, TimeUnit.SECONDS)); + factory.listener.onMessage( + factory.socket, "{\"type\":\"session.closed\",\"reason\":\"connection_lost\"}"); + assertNotNull(callback.errors.poll(5, TimeUnit.SECONDS)); + assertTrue(conversation.awaitSessionClosed(0)); + conversation.close(); + assertTrue(factory.socket.cancelCalls.get() > 0); + } + + @Test + void malformedMessagesReportErrorButLeaveConnectionUsable() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + for (String bad : new String[] {"{", "[]", "null", "{type:'bad'}"}) { + factory.listener.onMessage(factory.socket, bad); + assertNotNull(callback.errors.poll(5, TimeUnit.SECONDS)); + conversation.checkStatus(); + } + assertTrue(callback.events.isEmpty()); + } + + @Test + void missingOrNonStringTypeDoesNotInventState() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + for (String json : + new String[] { + "{}", + "{\"type\":null}", + "{\"type\":1}", + "{\"type\":\"session.created\",\"session\":null}", + "{\"type\":\"session.created\",\"session\":{}}" + }) { + factory.listener.onMessage(factory.socket, json); + assertEquals(JsonUtils.parse(json), callback.events.poll(5, TimeUnit.SECONDS)); + conversation.checkStatus(); + assertNull(conversation.getSessionId()); + } + assertTrue(callback.errors.isEmpty()); + } + + @Test + void sessionClosedUnblocksAnAlreadyWaitingThread() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> conversation.awaitSessionClosed(5)); + factory.listener.onMessage( + factory.socket, "{\"type\":\"session.closed\",\"reason\":\"remote_hangup\"}"); + assertTrue(result.get(5, TimeUnit.SECONDS)); + assertTrue(conversation.awaitSessionClosed(0)); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void serverErrorReachesUnifiedErrorCallbackAndKeepsOriginalEvent() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + JsonObject event = + JsonUtils.parse( + "{\"type\":\"error\",\"event_id\":\"server-event\",\"client_event_id\":\"sent-event\"," + + "\"request_id\":\"request-1\",\"error\":{\"type\":\"invalid_request_error\"," + + "\"code\":\"invalid_format\",\"message\":\"Use pcm\",\"param\":\"input_audio_format\"}}"); + factory.listener.onMessage(factory.socket, event.toString()); + + Throwable failure = callback.errors.poll(1, TimeUnit.SECONDS); + assertInstanceOf(ApiException.class, failure); + ApiException apiFailure = (ApiException) failure; + assertEquals("invalid_format", apiFailure.getStatus().getCode()); + assertEquals("Use pcm", apiFailure.getStatus().getMessage()); + assertEquals("request-1", apiFailure.getStatus().getRequestId()); + assertEquals(event, callback.events.poll(1, TimeUnit.SECONDS)); + conversation.checkStatus(); + } + + @Test + void throwingErrorCallbackDoesNotSuppressOriginalServerEvent() throws Exception { + FakeFactory factory = new FakeFactory(); + callback = + new RecordingCallback() { + @Override + public void onError(Throwable failure) { + errors.add(failure); + throw new IllegalStateException("error-handler-failed"); + } + }; + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + String event = "{\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\"}}"; + factory.listener.onMessage(factory.socket, event); + assertInstanceOf(ApiException.class, callback.errors.poll(1, TimeUnit.SECONDS)); + assertEquals(JsonUtils.parse(event), callback.events.poll(1, TimeUnit.SECONDS)); + conversation.checkStatus(); + } + + @Test + void invalidRequestErrorDoesNotTerminateSession() throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + for (String errorType : new String[] {"invalid_request_error", "server_error"}) { + String json = "{\"type\":\"error\",\"error\":{\"type\":\"" + errorType + "\"}}"; + factory.listener.onMessage(factory.socket, json); + assertEquals(JsonUtils.parse(json), callback.events.poll(5, TimeUnit.SECONDS)); + Throwable failure = callback.errors.poll(1, TimeUnit.SECONDS); + assertInstanceOf(ApiException.class, failure); + assertEquals(errorType, ((ApiException) failure).getStatus().getCode()); + conversation.checkStatus(); + } + } + + @ParameterizedTest + @MethodSource("incompleteServerErrors") + void incompleteServerErrorStillReachesBothCallbacks(String json) throws Exception { + FakeFactory factory = new FakeFactory(); + conversation = new QwenLiveConversation(param, callback, factory, 1); + conversation.connect(); + factory.listener.onMessage(factory.socket, json); + Throwable failure = callback.errors.poll(1, TimeUnit.SECONDS); + assertInstanceOf(ApiException.class, failure); + assertNotNull(((ApiException) failure).getStatus().getMessage()); + assertEquals(JsonUtils.parse(json), callback.events.poll(1, TimeUnit.SECONDS)); + conversation.checkStatus(); + } + + static Stream incompleteServerErrors() { + return Stream.of( + "{\"type\":\"error\"}", + "{\"type\":\"error\",\"error\":null}", + "{\"type\":\"error\",\"error\":\"unexpected\"}", + "{\"type\":\"error\",\"error\":{\"code\":[],\"message\":{}}}"); + } +} From 7c1ce24ffed3b303528e40b271718bdeff9c83ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=81=A5=E4=BB=99?= Date: Tue, 10 Nov 2026 16:33:23 +0800 Subject: [PATCH 2/2] feat(voice-enrollment): support base64 audio for voice cloning Add an audio field to VoiceEnrollmentParam so create_voice/update_voice can submit a base64 audio data URI (data:audio/wav;base64,...) in addition to the existing url; validate() now accepts either url or audio. Add createVoiceWithAudio/updateVoiceWithAudio overloads on VoiceEnrollmentService and pass customParam.audio through the existing methods. Existing url-based calls are unchanged. Covered by TestVoiceEnrollment. --- .../enrollment/VoiceEnrollmentParam.java | 23 ++- .../enrollment/VoiceEnrollmentService.java | 60 ++++++++ .../dashscope/TestVoiceEnrollment.java | 140 ++++++++++++++++++ 3 files changed, 217 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentParam.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentParam.java index e9a0036..2683b43 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentParam.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentParam.java @@ -20,6 +20,7 @@ public class VoiceEnrollmentParam extends HalfDuplexServiceParam { private String targetModel; private VoiceEnrollmentOperationType operationType; private String url; + private String audio; private String voiceId; private List languageHints = null; @@ -48,7 +49,12 @@ public JsonObject getInput() { input.addProperty(ApiKeywords.ACTION, operationType.getValue()); input.addProperty("target_model", targetModel); input.addProperty("prefix", prefix); - input.addProperty("url", url); + if (url != null) { + input.addProperty("url", url); + } + if (audio != null) { + input.addProperty("audio", audio); + } if (languageHints != null) { input.add("language_hints", JsonUtils.toJsonArray(languageHints)); } @@ -80,7 +86,12 @@ public JsonObject getInput() { case UPDATE: input.addProperty(ApiKeywords.ACTION, operationType.getValue()); input.addProperty("voice_id", voiceId); - input.addProperty("url", url); + if (url != null) { + input.addProperty("url", url); + } + if (audio != null) { + input.addProperty("audio", audio); + } break; case DELETE: input.addProperty(ApiKeywords.ACTION, operationType.getValue()); @@ -104,8 +115,8 @@ public void validate() throws InputRequiredException { } switch (operationType) { case CREATE: - if (url == null) { - throw new InputRequiredException("url is required when Create Voice"); + if (url == null && audio == null) { + throw new InputRequiredException("url or audio is required when Create Voice"); } if (prefix == null) { throw new InputRequiredException("prefix is required"); @@ -131,8 +142,8 @@ public void validate() throws InputRequiredException { if (voiceId == null) { throw new InputRequiredException("voice id is required when Update Voice"); } - if (url == null) { - throw new InputRequiredException("url is required when Update Voice"); + if (url == null && audio == null) { + throw new InputRequiredException("url or audio is required when Update Voice"); } break; case DELETE: diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentService.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentService.java index 23efe5d..60235e2 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentService.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/enrollment/VoiceEnrollmentService.java @@ -140,6 +140,38 @@ public Voice createVoice( .targetModel(targetModel) .prefix(prefix) .url(url) + .audio(customParam.getAudio()) + .apiKey(apikey) + .languageHints(customParam.getLanguageHints()) + .headers(customParam.getHeaders()) + .resources(customParam.getResources()) + .maxPromptAudioLength(customParam.getMaxPromptAudioLength()) + .parameters(customParam.getParameters()) + .workspace(customParam.getWorkspace()) + .build(); + param.validate(); + DashScopeResult dashScopeResult = syncApi.call(param); + lastRequestId = dashScopeResult.getRequestId(); + return Voice.voiceFromCreateResult(dashScopeResult); + } + + /** Create a cloned voice from a base64 audio data URI instead of a URL. */ + public Voice createVoiceWithAudio(String targetModel, String prefix, String audio) + throws NoApiKeyException, InputRequiredException { + return createVoiceWithAudio( + targetModel, prefix, audio, VoiceEnrollmentParam.builder().model(this.model).build()); + } + + public Voice createVoiceWithAudio( + String targetModel, String prefix, String audio, VoiceEnrollmentParam customParam) + throws NoApiKeyException, InputRequiredException { + VoiceEnrollmentParam param = + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.CREATE) + .model(customParam.getModel().isEmpty() ? this.model : customParam.getModel()) + .targetModel(targetModel) + .prefix(prefix) + .audio(audio) .apiKey(apikey) .languageHints(customParam.getLanguageHints()) .headers(customParam.getHeaders()) @@ -203,6 +235,34 @@ public void updateVoice(String voiceId, String url, VoiceEnrollmentParam customP .model(this.model) .voiceId(voiceId) .url(url) + .audio(customParam.getAudio()) + .apiKey(apikey) + .headers(customParam.getHeaders()) + .resources(customParam.getResources()) + .parameters(customParam.getParameters()) + .workspace(customParam.getWorkspace()) + .build(); + param.validate(); + DashScopeResult dashScopeResult = syncApi.call(param); + lastRequestId = dashScopeResult.getRequestId(); + } + + /** Update an existing voice with a base64 audio data URI instead of a URL. */ + public void updateVoiceWithAudio(String voiceId, String audio) + throws NoApiKeyException, InputRequiredException { + updateVoiceWithAudio( + voiceId, audio, VoiceEnrollmentParam.builder().model(this.model).build()); + } + + public void updateVoiceWithAudio( + String voiceId, String audio, VoiceEnrollmentParam customParam) + throws NoApiKeyException, InputRequiredException { + VoiceEnrollmentParam param = + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.UPDATE) + .model(this.model) + .voiceId(voiceId) + .audio(audio) .apiKey(apikey) .headers(customParam.getHeaders()) .resources(customParam.getResources()) diff --git a/src/test/java/com/alibaba/dashscope/TestVoiceEnrollment.java b/src/test/java/com/alibaba/dashscope/TestVoiceEnrollment.java index 6edc837..846c5f9 100644 --- a/src/test/java/com/alibaba/dashscope/TestVoiceEnrollment.java +++ b/src/test/java/com/alibaba/dashscope/TestVoiceEnrollment.java @@ -1,6 +1,13 @@ package com.alibaba.dashscope; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +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 com.alibaba.dashscope.audio.ttsv2.enrollment.Voice; +import com.alibaba.dashscope.audio.ttsv2.enrollment.VoiceEnrollmentOperationType; +import com.alibaba.dashscope.audio.ttsv2.enrollment.VoiceEnrollmentParam; import com.alibaba.dashscope.audio.ttsv2.enrollment.VoiceEnrollmentService; import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; @@ -8,6 +15,7 @@ import com.alibaba.dashscope.utils.JsonUtils; import com.google.gson.JsonObject; import java.io.IOException; +import java.util.concurrent.TimeUnit; import okhttp3.mockwebserver.MockResponse; import okhttp3.mockwebserver.MockWebServer; import org.junit.jupiter.api.AfterAll; @@ -65,4 +73,136 @@ public void testVoiceEnrollment() voice = service.queryVoice("voice-nuomai-0b84d031ca224adca6354ad45c6efcd0"); assert (voice.getResourceLink().equals("http://xxxx")); } + + @Test + public void createVoiceWithAudioSendsAudioField() throws Exception { + drainRequests(); + mockServer.enqueue( + TestUtils.createMockResponse( + JsonUtils.toJson( + JsonUtils.parse( + "{'output': {'voice_id': 'voice-myvoice-abc'}, 'usage': {}, 'request_id': 'r1'}")), + 200)); + VoiceEnrollmentService service = new VoiceEnrollmentService("1234"); + Voice voice = + service.createVoiceWithAudio( + "qwen-audio-3.0-tts-flash", "myvoice", "data:audio/wav;base64,AAAA"); + assertEquals("voice-myvoice-abc", voice.getVoiceId()); + JsonObject body = JsonUtils.parse(mockServer.takeRequest().getBody().readUtf8()); + assertEquals("voice-enrollment", body.get("model").getAsString()); + JsonObject input = body.getAsJsonObject("input"); + assertEquals("create_voice", input.get("action").getAsString()); + assertEquals("qwen-audio-3.0-tts-flash", input.get("target_model").getAsString()); + assertEquals("myvoice", input.get("prefix").getAsString()); + assertEquals("data:audio/wav;base64,AAAA", input.get("audio").getAsString()); + assertFalse(input.has("url")); + } + + @Test + public void createVoiceWithUrlStillSendsUrlAndNoAudio() throws Exception { + drainRequests(); + mockServer.enqueue( + TestUtils.createMockResponse( + JsonUtils.toJson( + JsonUtils.parse( + "{'output': {'voice_id': 'voice-nuomai-x'}, 'usage': {}, 'request_id': 'r2'}")), + 200)); + VoiceEnrollmentService service = new VoiceEnrollmentService("1234"); + service.createVoice("cosyvoice-v1", "nuomai", "https://xxxx"); + JsonObject input = + JsonUtils.parse(mockServer.takeRequest().getBody().readUtf8()).getAsJsonObject("input"); + assertEquals("create_voice", input.get("action").getAsString()); + assertEquals("https://xxxx", input.get("url").getAsString()); + assertFalse(input.has("audio")); + } + + @Test + public void updateVoiceWithAudioSendsAudioField() throws Exception { + drainRequests(); + mockServer.enqueue( + TestUtils.createMockResponse( + JsonUtils.toJson(JsonUtils.parse("{'output': {}, 'usage': {}, 'request_id': 'r3'}")), + 200)); + VoiceEnrollmentService service = new VoiceEnrollmentService("1234"); + service.updateVoiceWithAudio("voice-myvoice-abc", "data:audio/wav;base64,BBBB"); + JsonObject input = + JsonUtils.parse(mockServer.takeRequest().getBody().readUtf8()).getAsJsonObject("input"); + assertEquals("update_voice", input.get("action").getAsString()); + assertEquals("voice-myvoice-abc", input.get("voice_id").getAsString()); + assertEquals("data:audio/wav;base64,BBBB", input.get("audio").getAsString()); + assertFalse(input.has("url")); + } + + @Test + public void validateAcceptsUrlOrAudioAndRejectsNeither() { + assertDoesNotThrow( + () -> + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.CREATE) + .model("voice-enrollment") + .targetModel("m") + .prefix("p") + .audio("data:audio/wav;base64,AAAA") + .build() + .validate()); + assertDoesNotThrow( + () -> + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.CREATE) + .model("voice-enrollment") + .targetModel("m") + .prefix("p") + .url("https://x") + .build() + .validate()); + assertThrows( + InputRequiredException.class, + () -> + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.CREATE) + .model("voice-enrollment") + .targetModel("m") + .prefix("p") + .build() + .validate()); + assertDoesNotThrow( + () -> + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.UPDATE) + .model("voice-enrollment") + .voiceId("v") + .audio("data:audio/wav;base64,AAAA") + .build() + .validate()); + assertThrows( + InputRequiredException.class, + () -> + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.UPDATE) + .model("voice-enrollment") + .voiceId("v") + .build() + .validate()); + } + + @Test + public void getInputCreateWithAudioOmitsUrl() { + JsonObject input = + VoiceEnrollmentParam.builder() + .operationType(VoiceEnrollmentOperationType.CREATE) + .model("voice-enrollment") + .targetModel("m") + .prefix("p") + .audio("data:audio/wav;base64,AAAA") + .build() + .getInput(); + assertEquals("data:audio/wav;base64,AAAA", input.get("audio").getAsString()); + assertFalse(input.has("url")); + } + + private static void drainRequests() throws InterruptedException { + while (mockServer.takeRequest(0, TimeUnit.SECONDS) != null) { + // discard requests recorded by other tests to keep this assertion isolated + } + } }