diff --git a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java index f63950ac..9b375c97 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeCallback.java @@ -17,4 +17,13 @@ public void onOpen() {} /** Will be called when the connection is closed. */ public abstract void onClose(int code, String reason); + + /** + * Will be called when the websocket connection fails, e.g. network disconnected, connect timeout, + * or other io exceptions. Default implementation does nothing, override it to be notified of such + * errors. + * + * @param throwable the exception that caused the failure. + */ + public void onError(Throwable throwable) {} } diff --git a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java index 2b2dcfd6..53ab7ba3 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java @@ -350,7 +350,10 @@ private void sendMessage(ByteString message) { @Override public void onOpen(WebSocket webSocket, Response response) { isOpen.set(true); - connectLatch.get().countDown(); + CountDownLatch latch = connectLatch.get(); + if (latch != null && latch.getCount() > 0) { + latch.countDown(); + } log.debug("WebSocket opened"); callback.onOpen(); } @@ -410,15 +413,28 @@ public void onMessage(WebSocket webSocket, String text) { @Override public void onClosed(WebSocket webSocket, int code, String reason) { isOpen.set(false); - connectLatch.get().countDown(); + CountDownLatch latch = connectLatch.get(); + if (latch != null && latch.getCount() > 0) { + latch.countDown(); + } log.debug("WebSocket closed: " + code + ", " + reason); callback.onClose(code, reason); } @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { - connectLatch.get().countDown(); - log.error("WebSocket failed: " + t.getMessage()); + isOpen.set(false); + isClosed.set(true); + CountDownLatch cLatch = connectLatch.get(); + if (cLatch != null) { + cLatch.countDown(); + } + CountDownLatch dLatch = disconnectLatch.get(); + if (dLatch != null) { + dLatch.countDown(); + } + log.error("WebSocket failed: " + t.getMessage(), t); + callback.onError(t); } @Override diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java index 9a856ce7..bcd53688 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java @@ -239,7 +239,7 @@ public Flowable streamingCallAsFlowable(Flowable return duplexApi .duplexCall( StreamInputTtsParamWithStream.fromStreamInputTtsParam( - this.parameters, inputStream, preRequestId, false)) + this.parameters, inputStream, preRequestId, false, this.canceled)) .filter(item -> item.getEvent() != WebSocketEventType.TASK_STARTED.getValue()) .map(SpeechSynthesisResult::fromDashScopeResult) .filter(item -> !canceled.get()) @@ -292,7 +292,8 @@ public Flowable callAsFlowable(String text) }, BackpressureStrategy.BUFFER), preRequestId, - true)) + true, + this.canceled)) .filter(item -> item.getEvent() != WebSocketEventType.TASK_STARTED.getValue()) .map(SpeechSynthesisResult::fromDashScopeResult) .doOnNext( @@ -368,7 +369,7 @@ private void startStream(boolean enableSsml) { try { duplexApi.duplexCall( SpeechSynthesizer.StreamInputTtsParamWithStream.fromStreamInputTtsParam( - this.parameters, textFrames, preRequestId, enableSsml), + this.parameters, textFrames, preRequestId, enableSsml, this.canceled), new ResultCallback() { // private Sentence lastSentence = null; @@ -441,9 +442,6 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { log.debug("[TtsV2] onComplete"); - if (canceled.get()) { - return; - } synchronized (SpeechSynthesizer.this) { state = SpeechSynthesisState.IDLE; } @@ -802,11 +800,19 @@ private static class StreamInputTtsParamWithStream extends SpeechSynthesisParam @NonNull private Flowable textStream; + /** + * Shared reference to the outer {@link SpeechSynthesizer}'s canceled flag. When set to true + * before the finish-task message is sent, the finish-task message will carry + * payload.input.directive="cancel" to notify the server to discard remaining audio. + */ + private AtomicBoolean canceled; + public static StreamInputTtsParamWithStream fromStreamInputTtsParam( SpeechSynthesisParam param, Flowable textStream, String preRequestId, - boolean enableSsml) { + boolean enableSsml, + AtomicBoolean canceled) { return StreamInputTtsParamWithStream.builder() .headers(param.getHeaders()) .parameters(param.getParameters()) @@ -817,9 +823,15 @@ public static StreamInputTtsParamWithStream fromStreamInputTtsParam( .model(param.getModel()) .voice(param.getVoice()) .apiKey(param.getApiKey()) + .canceled(canceled) .build(); } + @Override + public String getDirective() { + return (canceled != null && canceled.get()) ? "cancel" : null; + } + @Override public Flowable getStreamingData() { return textStream diff --git a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java index 94acb336..1837a9a2 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java @@ -212,7 +212,11 @@ public void sendText(String text) { } public void stopSynthesizer() { - sendTaskMessage("finish-task", new JsonObject()); + JsonObject input = new JsonObject(); + if (canceled.get()) { + input.addProperty("directive", "cancel"); + } + sendTaskMessage("finish-task", input); } @Override @@ -413,6 +417,7 @@ private void startStream(boolean enableSsml) throws NoApiKeyException, Interrupt connect(); } else { startStreamTimeStamp = System.currentTimeMillis(); + canceled.set(false); } checkConnectStatus(); // check websocket connection， if socket is closed. diff --git a/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java b/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java index 345a9cdf..5442be8b 100644 --- a/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java +++ b/src/main/java/com/alibaba/dashscope/base/FullDuplexServiceParam.java @@ -45,4 +45,16 @@ public void putHeader(String key, String value) { headers.put(key, value); } } + + /** + * The directive to be carried in payload.input.directive when sending the finish-task message. + * Subclasses can override this method to instruct the server with special semantics (e.g. + * "cancel") when the finish-task message is sent. Returns null by default, meaning no directive + * will be added. + * + * @return the directive string, or null if none. + */ + public String getDirective() { + return null; + } } diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java index 4672c26c..2e119db8 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java @@ -396,22 +396,33 @@ public void sendHeartBeat() { */ public void requestToRespond( String type, String text, MultiModalRequestParam.UpdateParams updateParams) { - requestParamWithStream.clearParameters(); - MultiModalRequestParam.CustomInput customInput = - MultiModalRequestParam.CustomInput.builder() - .directive("RequestToRespond") - .dialogId(currentDialogId) - .type(type) - .text(text) - .build(); - requestParamWithStream.setCustomInput(customInput); - if (updateParams != null && updateParams.images != null) { - requestParamWithStream.setImages(updateParams.images); - } - if (updateParams != null && updateParams.bizParams != null) { - requestParamWithStream.setBizParams(updateParams.bizParams); + // Synchronize the whole "mutate requestParamWithStream + send" sequence on the same + // monitor used by sendTextFrame/sendAudioData/sendFinishTaskMessage. Without this, + // concurrent calls to requestToRespond/updateInfo could interleave their + // clearParameters()/setXxx() mutations before either one reaches sendTextFrame, + // causing one caller's payload to be sent with another caller's data (race condition). + // synchronized is reentrant, so the nested lock acquired inside sendTextFrame is safe. + synchronized (MultiModalDialog.this) { + requestParamWithStream.clearParameters(); + MultiModalRequestParam.CustomInput customInput = + MultiModalRequestParam.CustomInput.builder() + .directive("RequestToRespond") + .dialogId(currentDialogId) + .type(type) + .text(text) + .build(); + requestParamWithStream.setCustomInput(customInput); + if (updateParams != null && updateParams.images != null) { + requestParamWithStream.setImages(updateParams.images); + } + if (updateParams != null && updateParams.bizParams != null) { + requestParamWithStream.setBizParams(updateParams.bizParams); + } + if (updateParams != null && updateParams.parameters != null) { + requestParamWithStream.setExtraParameters(updateParams.parameters); + } + sendTextFrame("RequestToRespond"); } - sendTextFrame("RequestToRespond"); } /** @@ -420,29 +431,37 @@ public void requestToRespond( *

param: updateParams Update parameters */ public void updateInfo(MultiModalRequestParam.UpdateParams updateParams) { - requestParamWithStream.clearParameters(); - MultiModalRequestParam.CustomInput customInput = - MultiModalRequestParam.CustomInput.builder() - .directive("UpdateInfo") - .dialogId(currentDialogId) - .build(); - requestParamWithStream.setCustomInput(customInput); - if (updateParams != null && updateParams.clientInfo != null) { - requestParamWithStream.setClientInfo(updateParams.clientInfo); - } - if (updateParams != null && updateParams.bizParams != null) { - requestParamWithStream.setBizParams(updateParams.bizParams); - } - if (updateParams != null && updateParams.images != null) { - requestParamWithStream.setImages(updateParams.images); - } - if (updateParams != null && updateParams.upStream != null) { - requestParamWithStream.setUpStream(updateParams.upStream); - } - if (updateParams != null && updateParams.downStream != null) { - requestParamWithStream.setDownStream(updateParams.downStream); + // See requestToRespond() for why this whole block must be synchronized on the same + // monitor as sendTextFrame: mutation of requestParamWithStream and the subsequent send + // must be atomic with respect to concurrent requestToRespond/updateInfo callers. + synchronized (MultiModalDialog.this) { + requestParamWithStream.clearParameters(); + MultiModalRequestParam.CustomInput customInput = + MultiModalRequestParam.CustomInput.builder() + .directive("UpdateInfo") + .dialogId(currentDialogId) + .build(); + requestParamWithStream.setCustomInput(customInput); + if (updateParams != null && updateParams.clientInfo != null) { + requestParamWithStream.setClientInfo(updateParams.clientInfo); + } + if (updateParams != null && updateParams.bizParams != null) { + requestParamWithStream.setBizParams(updateParams.bizParams); + } + if (updateParams != null && updateParams.images != null) { + requestParamWithStream.setImages(updateParams.images); + } + if (updateParams != null && updateParams.upStream != null) { + requestParamWithStream.setUpStream(updateParams.upStream); + } + if (updateParams != null && updateParams.downStream != null) { + requestParamWithStream.setDownStream(updateParams.downStream); + } + if (updateParams != null && updateParams.parameters != null) { + requestParamWithStream.setExtraParameters(updateParams.parameters); + } + sendTextFrame("UpdateInfo"); } - sendTextFrame("UpdateInfo"); } /** diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java index fa75ebf0..1909b2e3 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalRequestParam.java @@ -25,6 +25,13 @@ public class MultiModalRequestParam extends FullDuplexServiceParam { private CustomInput customInput; private List images; private String taskId; + /** + * Extra custom payload.parameters entries, mutable at runtime (unlike the immutable {@code + * parameters} field inherited from {@link FullDuplexServiceParam}). Used by {@code + * requestToRespond}/{@code updateInfo} to let callers pass arbitrary key/value pairs into + * payload.parameters for a specific request. + */ + private Map extraParameters; @Builder public static class CustomInput { @@ -142,6 +149,7 @@ public void clearParameters() { customInput = null; images = null; dialogAttributes = null; + extraParameters = null; } @Builder @@ -151,6 +159,8 @@ public static class UpdateParams { List images; BizParams bizParams; ClientInfo clientInfo; + /** Custom key/value pairs merged directly into payload.parameters. */ + Map parameters; } @Override @@ -263,6 +273,9 @@ public Map getParameters() { if (this.parameters != null) { params.putAll(this.parameters); } + if (this.extraParameters != null) { + params.putAll(this.extraParameters); + } return params; } diff --git a/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java b/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java index c01dce08..11dae0c2 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java +++ b/src/main/java/com/alibaba/dashscope/protocol/FullDuplexRequest.java @@ -172,7 +172,10 @@ public JsonObject getFinishedTaskMessage(String taskId) { wsMessage.add(ApiKeywords.HEADER, header); JsonObject payload = new JsonObject(); JsonObject input = new JsonObject(); - if (serviceOption.getTask().equals("multimodal-generation")) { + String directive = param.getDirective(); + if (directive != null) { + input.addProperty("directive", directive); + } else if (serviceOption.getTask().equals("multimodal-generation")) { input.addProperty("directive", "Stop"); } payload.add("input", input);