Merge branch 'main' into dev/errors

Author: lzsweb

pom.xml

@@ -40,7 +40,7 @@
     <name>DashScope Java SDK</name>
     <groupId>com.alibaba</groupId>
     <artifactId>dashscope-sdk-java</artifactId>
-    <version>2.22.25</version>
+    <version>2.22.28</version>
 
     <properties>
         <maven.compiler.source>8</maven.compiler.source>

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) {}
 }

src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java

@@ -379,7 +379,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();
   }
@@ -439,15 +442,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

src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java

@@ -239,7 +239,7 @@ public Flowable<SpeechSynthesisResult> streamingCallAsFlowable(Flowable<String>
     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<SpeechSynthesisResult> 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<DashScopeResult>() {
             //                        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<TextStreamItem> 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<TextStreamItem> 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<Object> getStreamingData() {
       return textStream

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
@@ -423,6 +427,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.

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;
+  }
 }