diff --git a/pom.xml b/pom.xml index 6ec5f87f..f74b3719 100644 --- a/pom.xml +++ b/pom.xml @@ -40,7 +40,7 @@ DashScope Java SDK com.alibaba dashscope-sdk-java - 2.23.4 + 2.23.5 8 diff --git a/src/main/java/com/alibaba/dashscope/Version.java b/src/main/java/com/alibaba/dashscope/Version.java index ecb9a44c..ed90f43b 100644 --- a/src/main/java/com/alibaba/dashscope/Version.java +++ b/src/main/java/com/alibaba/dashscope/Version.java @@ -3,5 +3,5 @@ package com.alibaba.dashscope; public final class Version { - public static final String version = "2.23.4"; + public static final String version = "2.23.5"; } diff --git a/src/main/java/com/alibaba/dashscope/agentstudio/resource/AgentStudioEventStream.java b/src/main/java/com/alibaba/dashscope/agentstudio/resource/AgentStudioEventStream.java index 24651acf..02068df9 100644 --- a/src/main/java/com/alibaba/dashscope/agentstudio/resource/AgentStudioEventStream.java +++ b/src/main/java/com/alibaba/dashscope/agentstudio/resource/AgentStudioEventStream.java @@ -5,6 +5,7 @@ import com.alibaba.dashscope.agentstudio.message.Message; import com.alibaba.dashscope.common.Status; import com.alibaba.dashscope.exception.ApiException; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.google.gson.Gson; import com.google.gson.GsonBuilder; import com.google.gson.JsonObject; @@ -92,6 +93,7 @@ private static ApiException wrapFailure(Throwable t, Response response) { return null; } int code = response.code(); + String requestId = DashScopeHeaders.findRequestId(response.headers().toMultimap()); String body = ""; String errorCode = null; String errorMsg = null; @@ -100,24 +102,34 @@ private static ApiException wrapFailure(Throwable t, Response response) { body = rb.string(); } } catch (IOException e) { - log.debug("Failed to read SSE failure response body", e); + log.debug( + "Failed to read SSE failure response body, requestId={}", + DashScopeHeaders.requestIdOrUnknown(requestId), + e); } // Parse the agentstudio error envelope {type:error, error:{code,message}} // to surface the server's actual error code instead of a generic one. if (body != null && !body.isEmpty()) { try { JsonObject parsed = GSON.fromJson(body, JsonObject.class); + if (parsed != null && parsed.has("request_id")) { + requestId = parsed.get("request_id").getAsString(); + } if (parsed != null && parsed.has("error") && parsed.get("error").isJsonObject()) { JsonObject errObj = parsed.getAsJsonObject("error"); if (errObj.has("code")) errorCode = errObj.get("code").getAsString(); if (errObj.has("message")) errorMsg = errObj.get("message").getAsString(); } } catch (Exception e) { - log.debug("Failed to parse SSE error body as JSON", e); + log.debug( + "Failed to parse SSE error body as JSON, requestId={}", + DashScopeHeaders.requestIdOrUnknown(requestId), + e); } } Status status = Status.builder() + .requestId(requestId) .statusCode(code) .code( errorCode != null diff --git a/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGenerationResult.java b/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGenerationResult.java index 5ed5882d..68dfdae0 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGenerationResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/codegeneration/CodeGenerationResult.java @@ -33,7 +33,10 @@ public static CodeGenerationResult fromDashScopeResult(DashScopeResult dashScope JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), CodeGenerationOutput.class)); } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationResult.java b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationResult.java index cc270dba..c732c0cd 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/generation/GenerationResult.java @@ -43,7 +43,10 @@ public static GenerationResult fromDashScopeResult(DashScopeResult dashScopeResu JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), GenerationOutput.class)); } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationListResult.java b/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationListResult.java index 0c5ff9f3..a02a1dc2 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationListResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationListResult.java @@ -46,7 +46,10 @@ public static ImageGenerationListResult fromDashScopeResult(DashScopeResult dash rs.message = dashScopeResult.getMessage(); return rs; } else { - log.error("Result no output: {}", dashScopeResult); + log.error( + "Result no output: {}, requestId={}", + dashScopeResult, + dashScopeResult.getRequestIdOrUnknown()); } return null; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationResult.java b/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationResult.java index 69093b0f..90f5d615 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/imagegeneration/ImageGenerationResult.java @@ -41,7 +41,10 @@ public static ImageGenerationResult fromDashScopeResult(DashScopeResult dashScop JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), ImageGenerationOutput.class)); } else { - log.error("Result no output: {}", dashScopeResult); + log.error( + "Result no output: {}, requestId={}", + dashScopeResult, + dashScopeResult.getRequestIdOrUnknown()); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisListResult.java b/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisListResult.java index f513704e..e3520320 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisListResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisListResult.java @@ -46,7 +46,10 @@ public static ImageSynthesisListResult fromDashScopeResult(DashScopeResult dashS rs.message = dashScopeResult.getMessage(); return rs; } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return null; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisResult.java b/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisResult.java index 74a8a4fa..fc589657 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/imagesynthesis/ImageSynthesisResult.java @@ -46,7 +46,10 @@ public static ImageSynthesisResult fromDashScopeResult(DashScopeResult dashScope JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), ImageSynthesisOutput.class)); } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java index d0c2d87a..6b00e085 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java +++ b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversation.java @@ -167,6 +167,7 @@ public Flowable streamCall(MultiModalConversationP preprocessInput(param); return syncApi .streamCall(param) + .filter(MultiModalConversation::shouldEmitStreamResult) .map(MultiModalConversationResult::fromDashScopeResult) .map(result -> mergeSingleResponse(result, toMergeResponse)) .doOnComplete( @@ -214,6 +215,9 @@ public void streamCall( new ResultCallback() { @Override public void onEvent(DashScopeResult msg) { + if (!shouldEmitStreamResult(msg)) { + return; + } MultiModalConversationResult result = MultiModalConversationResult.fromDashScopeResult(msg); MultiModalConversationResult mergedResult = @@ -239,6 +243,27 @@ public void onError(Exception e) { }); } + /** + * The new gateway may emit usage/search/metadata SSE frames without a model output. They are + * successful auxiliary frames, not failed model results. Keep frames carrying an output or an + * explicit business error, and silently skip successful auxiliary frames. + */ + private static boolean shouldEmitStreamResult(DashScopeResult result) { + if (result.getOutput() != null) { + return true; + } + String code = result.getCode(); + boolean hasError = code != null && !code.isEmpty(); + if (!hasError) { + log.debug( + "Ignore streaming event without output: event={}, requestId={}, usagePresent={}", + result.getEvent(), + result.getRequestIdOrUnknown(), + result.getUsage() != null); + } + return hasError; + } + private void preprocessInput(MultiModalConversationParam param) throws NoApiKeyException, UploadFileException { boolean hasUpload = false; diff --git a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationResult.java b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationResult.java index 309ddc0f..b0f84c71 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/multimodalconversation/MultiModalConversationResult.java @@ -41,7 +41,10 @@ public static MultiModalConversationResult fromDashScopeResult(DashScopeResult d JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), MultiModalConversationOutput.class)); } else { - log.error("Result no output: {}", dashScopeResult); + log.error( + "Result no output: {}, requestId={}", + dashScopeResult, + dashScopeResult.getRequestIdOrUnknown()); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisListResult.java b/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisListResult.java index c5ecbc24..a24a2114 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisListResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisListResult.java @@ -45,7 +45,10 @@ public static VideoSynthesisListResult fromDashScopeResult(DashScopeResult dashS rs.message = dashScopeResult.getMessage(); return rs; } else { - log.error("Result no output: {}", dashScopeResult); + log.error( + "Result no output: {}, requestId={}", + dashScopeResult, + dashScopeResult.getRequestIdOrUnknown()); } return null; } diff --git a/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisResult.java b/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisResult.java index 4cf6a430..aca91c5f 100644 --- a/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisResult.java +++ b/src/main/java/com/alibaba/dashscope/aigc/videosynthesis/VideoSynthesisResult.java @@ -46,7 +46,10 @@ public static VideoSynthesisResult fromDashScopeResult(DashScopeResult dashScope (JsonObject) dashScopeResult.getOutput(), VideoSynthesisOutput.class); result.setOutput(outputTemp); } else { - log.error("Result no output: {}", dashScopeResult); + log.error( + "Result no output: {}, requestId={}", + dashScopeResult, + dashScopeResult.getRequestIdOrUnknown()); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/app/ApplicationResult.java b/src/main/java/com/alibaba/dashscope/app/ApplicationResult.java index 1d5260d4..0602e0aa 100644 --- a/src/main/java/com/alibaba/dashscope/app/ApplicationResult.java +++ b/src/main/java/com/alibaba/dashscope/app/ApplicationResult.java @@ -61,7 +61,10 @@ public static ApplicationResult fromDashScopeResult(DashScopeResult dashScopeRes JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), ApplicationOutput.class)); } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return result; diff --git a/src/main/java/com/alibaba/dashscope/audio/asr/recognition/Recognition.java b/src/main/java/com/alibaba/dashscope/audio/asr/recognition/Recognition.java index c6c811e5..41c3b004 100644 --- a/src/main/java/com/alibaba/dashscope/audio/asr/recognition/Recognition.java +++ b/src/main/java/com/alibaba/dashscope/audio/asr/recognition/Recognition.java @@ -10,6 +10,7 @@ import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import com.google.gson.Gson; @@ -59,6 +60,10 @@ private static class AsyncCmdBuffer { private String preRequestId = null; + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } + @SuperBuilder private static class RecognitionParamWithStream extends RecognitionParam { @@ -163,7 +168,10 @@ public Flowable streamCall( if (startStreamTimeStamp < 0) { startStreamTimeStamp = System.currentTimeMillis(); } - log.debug("send audio frame: " + buffer.remaining()); + log.debug( + "send audio frame: {}, requestId={}", + buffer.remaining(), + getRequestIdForLog()); }), preRequestId)) .doOnComplete( @@ -190,18 +198,24 @@ public Flowable streamCall( && result.getSentence().getText() != null && !result.getSentence().getText().isEmpty()) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "Recv Result: " - + result.getSentence().getText() - + ", isEnd: " - + result.isSentenceEnd()); + "Recv Result: {}, isEnd: {}, requestId={}", + result.getSentence().getText(), + result.isSentenceEnd(), + getRequestIdForLog()); }) .doOnComplete( () -> { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); }); } @@ -250,27 +264,30 @@ public void call(RecognitionParam param, ResultCallback callb @Override public void onEvent(DashScopeResult message) { RecognitionResult recognitionResult = RecognitionResult.fromDashScopeResult(message); + if (lastRequestId.get() == null && recognitionResult.getRequestId() != null) { + lastRequestId.set(recognitionResult.getRequestId()); + } if (recognitionResult.getSentence() != null && recognitionResult.getSentence().isHeartbeat()) { - log.debug("recv heartbeat"); + log.debug("recv heartbeat, requestId={}", getRequestIdForLog()); return; } - if (lastRequestId.get() == null && recognitionResult.getRequestId() != null) { - lastRequestId.set(recognitionResult.getRequestId()); - } if (!recognitionResult.isCompleteResult()) { if (firstPackageTimeStamp < 0 && recognitionResult.getSentence() != null && recognitionResult.getSentence().getText() != null && !recognitionResult.getSentence().getText().isEmpty()) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "Recv Result: " - + recognitionResult.getSentence().getText() - + ", isEnd: " - + recognitionResult.isSentenceEnd()); + "Recv Result: {}, isEnd: {}, requestId={}", + recognitionResult.getSentence().getText(), + recognitionResult.isSentenceEnd(), + getRequestIdForLog()); callback.onEvent(recognitionResult); } } @@ -278,7 +295,10 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); synchronized (Recognition.this) { state = RecognitionState.IDLE; } @@ -309,7 +329,7 @@ public void onError(Exception e) { stopLatch.get().countDown(); } } - log.debug("Recognition started"); + log.debug("Recognition started, requestId={}", getRequestIdForLog()); } public String call(RecognitionParam param, File file) { @@ -361,31 +381,37 @@ public String call(RecognitionParam param, File file) { .doOnComplete( () -> { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); }) .blockingSubscribe( res -> { RecognitionResult recognitionResult = RecognitionResult.fromDashScopeResult(res); - if (recognitionResult.getSentence().isHeartbeat()) { - log.debug("recv heartbeat"); - return; - } if (lastRequestId.get() == null && recognitionResult.getRequestId() != null) { lastRequestId.set(recognitionResult.getRequestId()); } + if (recognitionResult.getSentence().isHeartbeat()) { + log.debug("recv heartbeat, requestId={}", getRequestIdForLog()); + return; + } if (!recognitionResult.isCompleteResult() && recognitionResult.getSentence() != null && recognitionResult.getSentence().getText() != null && !recognitionResult.getSentence().getText().isEmpty()) { if (firstPackageTimeStamp < 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "Recv Result: " - + recognitionResult.getSentence().getText() - + ", isEnd: " - + recognitionResult.isSentenceEnd()); + "Recv Result: {}, isEnd: {}, requestId={}", + recognitionResult.getSentence().getText(), + recognitionResult.isSentenceEnd(), + getRequestIdForLog()); sentenceList.add(recognitionResult.getSentence()); } }, @@ -416,7 +442,7 @@ public void sendAudioFrame(ByteBuffer audioFrame) { if (this.startStreamTimeStamp < 0) { this.startStreamTimeStamp = System.currentTimeMillis(); } - log.debug("send audio frame: " + audioFrame.remaining()); + log.debug("send audio frame: {}, requestId={}", audioFrame.remaining(), getRequestIdForLog()); synchronized (this) { if (state != RecognitionState.RECOGNITION_STARTED) { throw new ApiException( diff --git a/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerChat.java b/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerChat.java index 0c74dca8..46939453 100644 --- a/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerChat.java +++ b/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerChat.java @@ -10,6 +10,7 @@ import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import io.reactivex.BackpressureStrategy; @@ -55,6 +56,11 @@ private static class AsyncCmdBuffer { private AtomicBoolean isSentenceEnd = new AtomicBoolean(false); private AtomicBoolean stopped = new AtomicBoolean(false); private String preRequestId = null; + private AtomicReference lastRequestId = new AtomicReference<>(null); + + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } @SuperBuilder private static class TranslationRecognizerParamWithStream extends TranslationRecognizerParam { @@ -197,18 +203,26 @@ public void call( public void onEvent(DashScopeResult message) { TranslationRecognizerResult translationRecognizerResult = TranslationRecognizerResult.fromDashScopeResult(message); + if (translationRecognizerResult.getRequestId() != null + && !translationRecognizerResult.getRequestId().isEmpty()) { + lastRequestId.set(translationRecognizerResult.getRequestId()); + } if (!translationRecognizerResult.isCompleteResult()) { if (firstPackageTimeStamp < 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "[Recv Result] transcription: " - + translationRecognizerResult.getTranslationResult() - + " translation: " - + translationRecognizerResult.getTranscriptionResult()); + "[Recv Result] transcription: {}, translation: {}, requestId={}", + translationRecognizerResult.getTranslationResult(), + translationRecognizerResult.getTranscriptionResult(), + getRequestIdForLog()); if (translationRecognizerResult.isSentenceEnd()) { - log.debug("[Chat] recv sentence end, stop asr"); + log.debug( + "[Chat] recv sentence end, stop asr, requestId={}", getRequestIdForLog()); isSentenceEnd.set(true); } callback.onEvent(translationRecognizerResult); @@ -218,7 +232,10 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); synchronized (TranslationRecognizerChat.this) { state = TranslationRecognizerState.IDLE; } @@ -249,7 +266,10 @@ public void onError(Exception e) { stopLatch.get().countDown(); } } - log.debug("TranslationRecognizerChat started, state is {}", state.getValue()); + log.debug( + "TranslationRecognizerChat started, state is {}, requestId={}", + state.getValue(), + getRequestIdForLog()); } /** @@ -263,7 +283,7 @@ public void onError(Exception e) { */ public boolean sendAudioFrame(ByteBuffer audioFrame) { if (isSentenceEnd.get()) { - log.debug("skip audio due to has sentence end"); + log.debug("skip audio due to has sentence end, requestId={}", getRequestIdForLog()); return false; } if (audioFrame == null) { @@ -272,7 +292,7 @@ public boolean sendAudioFrame(ByteBuffer audioFrame) { if (this.startStreamTimeStamp < 0) { this.startStreamTimeStamp = System.currentTimeMillis(); } - log.debug("send audio frame: " + audioFrame.remaining()); + log.debug("send audio frame: {}, requestId={}", audioFrame.remaining(), getRequestIdForLog()); synchronized (this) { if (state != TranslationRecognizerState.SPEECH_TRANSLATION_STARTED) { throw new ApiException( @@ -295,10 +315,12 @@ public boolean sendAudioFrame(ByteBuffer audioFrame) { public void stop() { this.stopStreamTimeStamp = System.currentTimeMillis(); if (stopped.get()) { - log.debug("TranslationRecognizerChat has already been stopped, skip"); + log.debug( + "TranslationRecognizerChat has already been stopped, skip, requestId={}", + getRequestIdForLog()); return; } - log.debug("stop TranslationRecognizerChat"); + log.debug("stop TranslationRecognizerChat, requestId={}", getRequestIdForLog()); synchronized (this) { if (state != TranslationRecognizerState.SPEECH_TRANSLATION_STARTED) { throw new ApiException( @@ -314,7 +336,7 @@ public void stop() { if (stopLatch.get() != null) { try { - log.debug("start waiting for task-finished"); + log.debug("start waiting for task-finished, requestId={}", getRequestIdForLog()); stopLatch.get().await(); } catch (InterruptedException ignored) { } @@ -333,6 +355,7 @@ private void reset() { this.isSentenceEnd.set(false); this.stopped.set(false); this.preRequestId = null; + this.lastRequestId.set(null); } /** First Package Delay is the time between start sending audio and receive first words package */ diff --git a/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerRealtime.java b/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerRealtime.java index 04cfb27c..9ccb8449 100644 --- a/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerRealtime.java +++ b/src/main/java/com/alibaba/dashscope/audio/asr/translation/TranslationRecognizerRealtime.java @@ -11,6 +11,7 @@ import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import io.reactivex.BackpressureStrategy; @@ -57,6 +58,11 @@ private static class AsyncCmdBuffer { private long stopStreamTimeStamp = -1; private long onCompleteTimeStamp = -1; private String preRequestId = null; + private AtomicReference lastRequestId = new AtomicReference<>(null); + + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } @SuperBuilder private static class TranslationRecognizerParamWithStream extends TranslationRecognizerParam { @@ -164,7 +170,10 @@ public Flowable streamCall( if (startStreamTimeStamp < 0) { startStreamTimeStamp = System.currentTimeMillis(); } - log.debug("send audio frame: " + buffer.remaining()); + log.debug( + "send audio frame: {}, requestId={}", + buffer.remaining(), + getRequestIdForLog()); }), preRequestId)) .doOnComplete( @@ -183,20 +192,29 @@ public Flowable streamCall( && !item.isCompleteResult()) .doOnNext( result -> { + if (result.getRequestId() != null && !result.getRequestId().isEmpty()) { + lastRequestId.set(result.getRequestId()); + } if (firstPackageTimeStamp < 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "[Recv Result] transcription: " - + result.getTranslationResult() - + " translation: " - + result.getTranscriptionResult()); + "[Recv Result] transcription: {}, translation: {}, requestId={}", + result.getTranslationResult(), + result.getTranscriptionResult(), + getRequestIdForLog()); }) .doOnComplete( () -> { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); }); } @@ -256,16 +274,23 @@ public void call( public void onEvent(DashScopeResult message) { TranslationRecognizerResult translationRecognizerResult = TranslationRecognizerResult.fromDashScopeResult(message); + if (translationRecognizerResult.getRequestId() != null + && !translationRecognizerResult.getRequestId().isEmpty()) { + lastRequestId.set(translationRecognizerResult.getRequestId()); + } if (!translationRecognizerResult.isCompleteResult()) { if (firstPackageTimeStamp < 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "[Recv Result] transcription: " - + translationRecognizerResult.getTranslationResult() - + " translation: " - + translationRecognizerResult.getTranscriptionResult()); + "[Recv Result] transcription: {}, translation: {}, requestId={}", + translationRecognizerResult.getTranslationResult(), + translationRecognizerResult.getTranscriptionResult(), + getRequestIdForLog()); callback.onEvent(translationRecognizerResult); } } @@ -273,7 +298,10 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); synchronized (TranslationRecognizerRealtime.this) { state = TranslationRecognizerState.IDLE; } @@ -304,7 +332,7 @@ public void onError(Exception e) { stopLatch.get().countDown(); } } - log.debug("TranslationRecognizerRealtime started"); + log.debug("TranslationRecognizerRealtime started, requestId={}", getRequestIdForLog()); } /** @@ -362,23 +390,33 @@ public TranslationRecognizerResultPack call(TranslationRecognizerParam param, Fi .doOnComplete( () -> { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); }) .blockingSubscribe( res -> { TranslationRecognizerResult translationRecognizerResult = TranslationRecognizerResult.fromDashScopeResult(res); + if (translationRecognizerResult.getRequestId() != null + && !translationRecognizerResult.getRequestId().isEmpty()) { + lastRequestId.set(translationRecognizerResult.getRequestId()); + } if (!translationRecognizerResult.isCompleteResult() && translationRecognizerResult.isSentenceEnd()) { if (firstPackageTimeStamp < 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("first package delay: " + getFirstPackageDelay()); + log.debug( + "first package delay: {}, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } log.debug( - "[Recv SentenceEnd]: transcription" - + translationRecognizerResult.getTranslationResult() - + ", translation: " - + translationRecognizerResult.getTranscriptionResult()); + "[Recv SentenceEnd]: transcription {}, translation: {}, requestId={}", + translationRecognizerResult.getTranslationResult(), + translationRecognizerResult.getTranscriptionResult(), + getRequestIdForLog()); results.setRequestId(translationRecognizerResult.getRequestId()); results .getTranslationResultList() @@ -418,7 +456,7 @@ public void sendAudioFrame(ByteBuffer audioFrame) { if (this.startStreamTimeStamp < 0) { this.startStreamTimeStamp = System.currentTimeMillis(); } - log.debug("send audio frame: " + audioFrame.remaining()); + log.debug("send audio frame: {}, requestId={}", audioFrame.remaining(), getRequestIdForLog()); synchronized (this) { if (state != TranslationRecognizerState.SPEECH_TRANSLATION_STARTED) { throw new ApiException( @@ -471,6 +509,7 @@ private void reset() { this.stopStreamTimeStamp = -1; this.onCompleteTimeStamp = -1; preRequestId = null; + lastRequestId.set(null); } /** First Package Delay is the time between start sending audio and receive first words package */ diff --git a/src/main/java/com/alibaba/dashscope/audio/http_tts/HttpSpeechSynthesizer.java b/src/main/java/com/alibaba/dashscope/audio/http_tts/HttpSpeechSynthesizer.java index 2dde1ac7..1f47ee03 100644 --- a/src/main/java/com/alibaba/dashscope/audio/http_tts/HttpSpeechSynthesizer.java +++ b/src/main/java/com/alibaba/dashscope/audio/http_tts/HttpSpeechSynthesizer.java @@ -10,6 +10,7 @@ import com.alibaba.dashscope.exception.InputRequiredException; import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.HttpMethod; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; @@ -38,6 +39,20 @@ public class HttpSpeechSynthesizer { /** Creates a new HttpSpeechSynthesizer instance with default settings. */ public HttpSpeechSynthesizer() {} + + private static String getRequestIdForLog(String requestId, Exception error) { + if (requestId != null && !requestId.isEmpty()) { + return requestId; + } + if (error instanceof ApiException) { + Status status = ((ApiException) error).getStatus(); + if (status != null) { + return DashScopeHeaders.requestIdOrUnknown(status.getRequestId()); + } + } + return "unknown"; + } + /** Creates a per-request ApiServiceOption with the specified SSE setting. */ private SynchronizeHalfDuplexApi createApi(boolean isSSE) { ApiServiceOption serviceOption = @@ -87,11 +102,15 @@ public HttpSpeechSynthesisResult call(HttpSpeechSynthesisParam param) throws ApiException, NoApiKeyException, InputRequiredException { param.validate(); SynchronizeHalfDuplexApi api = createApi(false); + DashScopeResult result = null; try { - DashScopeResult result = api.call(param); + result = api.call(param); return convertNonSSEResult(result); } catch (Exception e) { - log.error("Synchronous speech synthesis failed", e); + log.error( + "Synchronous speech synthesis failed, requestId={}", + getRequestIdForLog(result == null ? null : result.getRequestId(), e), + e); throw new ApiException(e); } } @@ -111,6 +130,7 @@ public void streamCall( throws ApiException, NoApiKeyException, InputRequiredException { param.validate(); SynchronizeHalfDuplexApi api = createApi(true); + AtomicReference requestId = new AtomicReference<>(null); try { api.streamCall( @@ -118,13 +138,19 @@ public void streamCall( new ResultCallback() { @Override public void onEvent(DashScopeResult message) { + if (message.getRequestId() != null && !message.getRequestId().isEmpty()) { + requestId.set(message.getRequestId()); + } try { HttpSpeechSynthesisResult result = convertResult(message); if (callback != null) { callback.onEvent(result); } } catch (Exception e) { - log.error("Failed to process audio result", e); + log.error( + "Failed to process audio result, requestId={}", + DashScopeHeaders.requestIdOrUnknown(requestId.get()), + e); if (callback != null) { callback.onError(e); } @@ -146,7 +172,7 @@ public void onError(Exception e) { } }); } catch (Exception e) { - log.error("Streaming call failed", e); + log.error("Streaming call failed, requestId={}", getRequestIdForLog(requestId.get(), e), e); throw new ApiException(e); } } @@ -167,12 +193,16 @@ public ByteBuffer callAndReturnAudio(HttpSpeechSynthesisParam param) ByteArrayOutputStream audioBuffer = new ByteArrayOutputStream(); CountDownLatch latch = new CountDownLatch(1); AtomicReference callbackError = new AtomicReference<>(); + AtomicReference requestId = new AtomicReference<>(null); streamCall( param, new ResultCallback() { @Override public void onEvent(HttpSpeechSynthesisResult result) { + if (result.getRequestId() != null && !result.getRequestId().isEmpty()) { + requestId.set(result.getRequestId()); + } try { if (result.getAudioData() != null) { audioBuffer.write(result.getAudioData()); @@ -216,7 +246,8 @@ public void onError(Exception e) { } catch (ApiException e) { throw e; } catch (Exception e) { - log.error("Failed to get audio data", e); + log.error( + "Failed to get audio data, requestId={}", getRequestIdForLog(requestId.get(), e), e); throw new ApiException(e); } } @@ -256,7 +287,10 @@ private HttpSpeechSynthesisResult convertResult(DashScopeResult dashScopeResult) dashScopeResult.getUsage().getAsJsonObject(), SpeechSynthesisUsage.class); result.setUsage(usage); } catch (Exception e) { - log.debug("Failed to parse usage information", e); + log.debug( + "Failed to parse usage information, requestId={}", + dashScopeResult.getRequestIdOrUnknown(), + e); } } @@ -290,7 +324,10 @@ private byte[] extractAudioData(DashScopeResult result) { try { return Base64.getDecoder().decode(audioBase64); } catch (IllegalArgumentException e) { - log.warn("Failed to decode Base64 audio data", e); + log.warn( + "Failed to decode Base64 audio data, requestId={}", + result.getRequestIdOrUnknown(), + e); } } } else if (audioElement.isJsonObject()) { @@ -302,7 +339,10 @@ private byte[] extractAudioData(DashScopeResult result) { try { return Base64.getDecoder().decode(audioBase64); } catch (IllegalArgumentException e) { - log.warn("Failed to decode Base64 audio data from audio.data", e); + log.warn( + "Failed to decode Base64 audio data from audio.data, requestId={}", + result.getRequestIdOrUnknown(), + e); } } } @@ -316,7 +356,10 @@ private byte[] extractAudioData(DashScopeResult result) { try { return Base64.getDecoder().decode(binaryBase64); } catch (IllegalArgumentException e) { - log.warn("Failed to decode Base64 binary data", e); + log.warn( + "Failed to decode Base64 binary data, requestId={}", + result.getRequestIdOrUnknown(), + e); } } } @@ -397,7 +440,10 @@ private HttpSpeechSynthesisResult convertNonSSEResult(DashScopeResult dashScopeR dashScopeResult.getUsage().getAsJsonObject(), SpeechSynthesisUsage.class); result.setUsage(usage); } catch (Exception e) { - log.debug("Failed to parse usage information", e); + log.debug( + "Failed to parse usage information, requestId={}", + dashScopeResult.getRequestIdOrUnknown(), + e); } } 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 0392302a..913d7232 100644 --- a/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java +++ b/src/main/java/com/alibaba/dashscope/audio/omni/OmniRealtimeConversation.java @@ -40,6 +40,20 @@ public class OmniRealtimeConversation extends WebSocketListener { private long lastFirstTextDelay = -1; private AtomicBoolean isClosed = new AtomicBoolean(false); private final AtomicReference disconnectLatch = new AtomicReference<>(null); + private volatile String requestId; + + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(requestId); + } + + private void updateRequestIdFromResponse(Response response) { + if (response != null && (requestId == null || requestId.isEmpty())) { + String responseRequestId = DashScopeHeaders.findRequestId(response.headers().toMultimap()); + if (responseRequestId != null && !responseRequestId.isEmpty()) { + requestId = responseRequestId; + } + } + } /** * Constructor @@ -160,7 +174,11 @@ public void appendAudio(String audioBase64) { OmniRealtimeConstants.PROTOCOL_TYPE, OmniRealtimeConstants.PROTOCOL_EVENT_TYPE_APPEND_AUDIO); append_request.put(OmniRealtimeConstants.PROTOCOL_AUDIO, audioBase64); - log.debug("append audio with eid: {}, length: {}", event_id, audioBase64.length()); + log.debug( + "append audio with eid: {}, length: {}, requestId={}", + event_id, + audioBase64.length(), + getRequestIdForLog()); sendMessage(createGson().toJson(append_request), false); } @@ -178,7 +196,11 @@ public void appendVideo(String videoBase64) { OmniRealtimeConstants.PROTOCOL_TYPE, OmniRealtimeConstants.PROTOCOL_EVENT_TYPE_APPEND_VIDEO); append_request.put(OmniRealtimeConstants.PROTOCOL_VIDEO, videoBase64); - log.debug("append video with eid: " + event_id + ", length: " + videoBase64.length()); + log.debug( + "append video with eid: {}, length: {}, requestId={}", + event_id, + videoBase64.length(), + getRequestIdForLog()); sendMessage(createGson().toJson(append_request), false); } @@ -337,7 +359,7 @@ private Gson createGson() { private void sendMessage(String message, boolean enableLog) { if (enableLog == true) { - log.debug("send message: " + message); + log.debug("send message: {}, requestId={}", message, getRequestIdForLog()); } Boolean isOk = websocktetClient.send(message); } @@ -349,21 +371,23 @@ private void sendMessage(ByteString message) { /** WebSocket callbacks */ @Override public void onOpen(WebSocket webSocket, Response response) { + updateRequestIdFromResponse(response); isOpen.set(true); CountDownLatch latch = connectLatch.get(); if (latch != null && latch.getCount() > 0) { latch.countDown(); } - log.debug("WebSocket opened"); + log.debug("WebSocket opened, requestId={}", getRequestIdForLog()); callback.onOpen(); } @Override public void onMessage(WebSocket webSocket, String text) { if (text.length() > 1024) { - log.debug("Received message: " + text.substring(0, 1024)); + log.debug( + "Received message: {}, requestId={}", text.substring(0, 1024), getRequestIdForLog()); } else { - log.debug("Received message: " + text); + log.debug("Received message: {}, requestId={}", text, getRequestIdForLog()); } JsonObject response = JsonUtils.parse(text); callback.onEvent(response); @@ -391,16 +415,14 @@ public void onMessage(WebSocket webSocket, String text) { break; case OmniRealtimeConstants.PROTOCOL_RESPONSE_TYPE_RESPONSE_DONE: log.debug( - "[Metric] response: " - + lastResponseId - + ", first text delay: " - + lastFirstTextDelay - + " ms, first audio delay: " - + lastFirstAudioDelay - + " ms"); + "[Metric] response: {}, first text delay: {} ms, first audio delay: {} ms, requestId={}", + lastResponseId, + lastFirstTextDelay, + lastFirstAudioDelay, + getRequestIdForLog()); break; case OmniRealtimeConstants.PROTOCOL_RESPONSE_TYPE_SESSION_FINISHED: - log.info("session: " + sessionId + " finished"); + log.info("session: {} finished, requestId={}", sessionId, getRequestIdForLog()); CountDownLatch latch = disconnectLatch.get(); if (latch != null) { latch.countDown(); @@ -417,12 +439,13 @@ public void onClosed(WebSocket webSocket, int code, String reason) { if (latch != null && latch.getCount() > 0) { latch.countDown(); } - log.debug("WebSocket closed: " + code + ", " + reason); + log.debug("WebSocket closed: {}, {}, requestId={}", code, reason, getRequestIdForLog()); callback.onClose(code, reason); } @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { + updateRequestIdFromResponse(response); isOpen.set(false); isClosed.set(true); CountDownLatch cLatch = connectLatch.get(); @@ -433,7 +456,7 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) { if (dLatch != null) { dLatch.countDown(); } - log.error("WebSocket failed: " + t.getMessage(), t); + log.error("WebSocket failed: {}, requestId={}", t.getMessage(), getRequestIdForLog(), t); callback.onError(t); } @@ -441,6 +464,6 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) { public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) { isClosed.set(true); websocktetClient.close(code, reason); - log.debug("WebSocket closing: " + code + ", " + reason); + log.debug("WebSocket closing: {}, {}, requestId={}", code, reason, getRequestIdForLog()); } } diff --git a/src/main/java/com/alibaba/dashscope/audio/protocol/AudioWebsocketRequest.java b/src/main/java/com/alibaba/dashscope/audio/protocol/AudioWebsocketRequest.java index 69375e88..14e8cf33 100644 --- a/src/main/java/com/alibaba/dashscope/audio/protocol/AudioWebsocketRequest.java +++ b/src/main/java/com/alibaba/dashscope/audio/protocol/AudioWebsocketRequest.java @@ -5,6 +5,8 @@ import com.alibaba.dashscope.protocol.okhttp.OkHttpClientFactory; import com.alibaba.dashscope.utils.ApiKey; import com.alibaba.dashscope.utils.Constants; +import com.alibaba.dashscope.utils.JsonUtils; +import com.google.gson.JsonObject; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -20,12 +22,47 @@ public class AudioWebsocketRequest extends WebSocketListener { private OkHttpClient client; private WebSocket websocketClient; + private volatile String requestId; private AtomicBoolean isOpen = new AtomicBoolean(false); private AtomicReference connectLatch = new AtomicReference<>(null); private AtomicBoolean isClosed = new AtomicBoolean(false); private AudioWebsocketCallback callback; private Integer connectTimeout = 5000; + private String getRequestIdOrUnknown() { + return DashScopeHeaders.requestIdOrUnknown(requestId); + } + + private void updateRequestIdFromMessage(String message) { + try { + JsonObject jsonMessage = JsonUtils.fromJson(message, JsonObject.class); + if (jsonMessage == null || !jsonMessage.has("header")) { + return; + } + JsonObject header = jsonMessage.getAsJsonObject("header"); + if (header != null && header.has("task_id") && !header.get("task_id").isJsonNull()) { + String messageRequestId = header.get("task_id").getAsString(); + if (messageRequestId != null && !messageRequestId.isEmpty()) { + this.requestId = messageRequestId; + } + } + } catch (RuntimeException ignored) { + // Preserve support for non-JSON messages; the original message handling reports any error. + } + } + + private void updateRequestIdFromResponse(Response response) { + if (response != null && (requestId == null || requestId.isEmpty())) { + updateRequestId(DashScopeHeaders.findRequestId(response.headers().toMultimap())); + } + } + + private void updateRequestId(String requestId) { + if (requestId != null && !requestId.isEmpty()) { + this.requestId = requestId; + } + } + public boolean isOpen() { return isOpen.get(); } @@ -84,11 +121,14 @@ private Request buildConnectionRequest( private void sendMessage(String message, boolean enableLog) { checkStatus(); + updateRequestIdFromMessage(message); if (enableLog) { - log.debug("send message: " + message); + log.debug("send message: {}, requestId={}", message, getRequestIdOrUnknown()); } if (!websocketClient.send(message)) { - log.warn("Failed to enqueue websocket text message for sending."); + log.warn( + "Failed to enqueue websocket text message for sending. requestId={}", + getRequestIdOrUnknown()); } } @@ -110,29 +150,33 @@ public void sendTextMessage(String message) { public void sendBinaryMessage(ByteString rawData) { checkStatus(); if (!websocketClient.send(rawData)) { - log.warn("Failed to enqueue websocket binary message for sending."); + log.warn( + "Failed to enqueue websocket binary message for sending. requestId={}", + getRequestIdOrUnknown()); } } @Override public void onOpen(WebSocket webSocket, Response response) { + updateRequestIdFromResponse(response); isOpen.set(true); if (connectLatch.get() != null) { connectLatch.get().countDown(); } - log.debug("WebSocket opened"); + log.debug("WebSocket opened, requestId={}", getRequestIdOrUnknown()); callback.onOpen(); } @Override public void onMessage(WebSocket webSocket, String text) { + updateRequestIdFromMessage(text); callback.onMessage(webSocket, text); } @Override public void onMessage(WebSocket webSocket, ByteString bytes) { - log.debug("Received binary message"); + log.debug("Received binary message, requestId={}", getRequestIdOrUnknown()); callback.onMessage(webSocket, bytes.asByteBuffer()); } @@ -143,13 +187,14 @@ public void onClosed(WebSocket webSocket, int code, String reason) { if (connectLatch.get() != null) { connectLatch.get().countDown(); } - log.debug("WebSocket closed"); + log.debug("WebSocket closed, requestId={}", getRequestIdOrUnknown()); callback.onClose(code, reason); } @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { - log.error("WebSocket failed: " + t); + updateRequestIdFromResponse(response); + log.error("WebSocket failed: {}, requestId={}", t, getRequestIdOrUnknown()); if (connectLatch.get() != null) { connectLatch.get().countDown(); } diff --git a/src/main/java/com/alibaba/dashscope/audio/qwen_tts_realtime/QwenTtsRealtime.java b/src/main/java/com/alibaba/dashscope/audio/qwen_tts_realtime/QwenTtsRealtime.java index ced782fe..d43beced 100644 --- a/src/main/java/com/alibaba/dashscope/audio/qwen_tts_realtime/QwenTtsRealtime.java +++ b/src/main/java/com/alibaba/dashscope/audio/qwen_tts_realtime/QwenTtsRealtime.java @@ -36,6 +36,20 @@ public class QwenTtsRealtime extends WebSocketListener { private long lastFirstTextTime = -1; private long lastFirstAudioDelay = -1; private AtomicBoolean isClosed = new AtomicBoolean(false); + private volatile String requestId; + + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(requestId); + } + + private void updateRequestIdFromResponse(Response response) { + if (response != null && (requestId == null || requestId.isEmpty())) { + String responseRequestId = DashScopeHeaders.findRequestId(response.headers().toMultimap()); + if (responseRequestId != null && !responseRequestId.isEmpty()) { + requestId = responseRequestId; + } + } + } /** * Constructor @@ -236,7 +250,7 @@ private String generateSessionId() { private void sendMessage(String message, boolean enableLog) { if (enableLog == true) { - log.debug("send message: " + message); + log.debug("send message: {}, requestId={}", message, getRequestIdForLog()); } Boolean isOk = websocktetClient.send(message); } @@ -248,18 +262,20 @@ private void sendMessage(ByteString message) { /** WebSocket callbacks */ @Override public void onOpen(WebSocket webSocket, Response response) { + updateRequestIdFromResponse(response); isOpen.set(true); connectLatch.get().countDown(); - log.debug("WebSocket opened"); + log.debug("WebSocket opened, requestId={}", getRequestIdForLog()); callback.onOpen(); } @Override public void onMessage(WebSocket webSocket, String text) { if (text.length() > 1024) { - log.debug("Received message: " + text.substring(0, 1024)); + log.debug( + "Received message: {}, requestId={}", text.substring(0, 1024), getRequestIdForLog()); } else { - log.debug("Received message: " + text); + log.debug("Received message: {}, requestId={}", text, getRequestIdForLog()); } JsonObject response = JsonUtils.parse(text); callback.onEvent(response); @@ -279,11 +295,10 @@ public void onMessage(WebSocket webSocket, String text) { break; case QwenTtsRealtimeConstants.PROTOCOL_RESPONSE_TYPE_RESPONSE_DONE: log.debug( - "[Metric] response: " - + lastResponseId - + ", first audio delay: " - + lastFirstAudioDelay - + " ms"); + "[Metric] response: {}, first audio delay: {} ms, requestId={}", + lastResponseId, + lastFirstAudioDelay, + getRequestIdForLog()); break; } } @@ -293,7 +308,7 @@ public void onMessage(WebSocket webSocket, String text) { public void onClosed(WebSocket webSocket, int code, String reason) { isOpen.set(false); connectLatch.get().countDown(); - log.debug("WebSocket closed"); + log.debug("WebSocket closed, requestId={}", getRequestIdForLog()); callback.onClose(code, reason); } @@ -301,12 +316,13 @@ public void onClosed(WebSocket webSocket, int code, String reason) { public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) { isClosed.set(true); websocktetClient.close(code, reason); - log.debug("WebSocket closing: " + code + ", " + reason); + log.debug("WebSocket closing: {}, {}, requestId={}", code, reason, getRequestIdForLog()); } @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { + updateRequestIdFromResponse(response); connectLatch.get().countDown(); - log.error("WebSocket failed: " + t.getMessage()); + log.error("WebSocket failed: {}, requestId={}", t.getMessage(), getRequestIdForLog()); } } diff --git a/src/main/java/com/alibaba/dashscope/audio/tts/SpeechSynthesizer.java b/src/main/java/com/alibaba/dashscope/audio/tts/SpeechSynthesizer.java index 686d078c..21bec24d 100644 --- a/src/main/java/com/alibaba/dashscope/audio/tts/SpeechSynthesizer.java +++ b/src/main/java/com/alibaba/dashscope/audio/tts/SpeechSynthesizer.java @@ -8,6 +8,7 @@ import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import io.reactivex.Flowable; @@ -37,6 +38,10 @@ public final class SpeechSynthesizer { private String preRequestId = null; + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } + public String getLastRequestId() { return preRequestId; } @@ -61,6 +66,7 @@ public void call(SpeechSynthesisParam param, ResultCallback parameters = new HashMap<>(); parameters.put("pre_task_id", preRequestId); param.setParameters(parameters); @@ -79,6 +85,9 @@ public void onOpen(Status status) { @Override public void onEvent(DashScopeResult message) { + if (message.getRequestId() != null && !message.getRequestId().isEmpty()) { + lastRequestId.set(message.getRequestId()); + } SpeechSynthesisResult result = SpeechSynthesisResult.fromDashScopeResult(message); try { if (result.getTimestamp() != null) { @@ -98,7 +107,10 @@ public void onEvent(DashScopeResult message) { if (result.getAudioFrame() != null) { if (recvAudioLength == 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("[TtsV2] first package delay: " + getFirstPackageDelay() + " ms"); + log.debug( + "[TtsV2] first package delay: {} ms, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } recvAudioLength += (double) result.getAudioFrame().capacity() @@ -106,20 +118,28 @@ public void onEvent(DashScopeResult message) { long current = System.currentTimeMillis(); double current_rtf = (current - startStreamTimeStamp) / recvAudioLength; log.debug( - "[TtsV2] Recv Audio Binary: " - + result.getAudioFrame().capacity() - + " bytes, total audio " - + recvAudioLength - + " ms, current_rtf: " - + current_rtf); + "[TtsV2] Recv Audio Binary: {} bytes, total audio {} ms, current_rtf: {}, requestId={}", + result.getAudioFrame().capacity(), + recvAudioLength, + current_rtf, + getRequestIdForLog()); try { channel.write(result.getAudioFrame()); } catch (IOException e) { - log.error("Failed to write audio: {}", result.getAudioFrame(), e); + log.error( + "Failed to write audio: {}, requestId={}", + result.getAudioFrame(), + getRequestIdForLog(), + e); } } } catch (Exception e) { - log.error("Failed to parse response: {}", message, e); + log.error( + "Failed to parse response, event: {}, output: {}, requestId={}", + message.getEvent(), + message.getOutput(), + getRequestIdForLog(), + e); callback.onError(e); } callback.onEvent(result); @@ -132,7 +152,7 @@ public void onComplete() { channel.close(); outputStream.close(); } catch (IOException e) { - log.error("Failed to close channel: {}", channel, e); + log.error("Failed to close channel: {}, requestId={}", channel, getRequestIdForLog(), e); } callback.onComplete(); } @@ -154,6 +174,7 @@ public Flowable streamCall(SpeechSynthesisParam param) { recvAudioLength = 0; firstPackageTimeStamp = -1; preRequestId = UUID.randomUUID().toString(); + lastRequestId.set(null); Map parameters = new HashMap<>(); parameters.put("pre_task_id", preRequestId); param.setParameters(parameters); @@ -167,6 +188,9 @@ public Flowable streamCall(SpeechSynthesisParam param) { .streamCall(param) .map( message -> { + if (message.getRequestId() != null && !message.getRequestId().isEmpty()) { + lastRequestId.set(message.getRequestId()); + } SpeechSynthesisResult result = SpeechSynthesisResult.fromDashScopeResult(message); if (result.getTimestamp() != null) { Sentence sentence = result.getTimestamp(); @@ -185,7 +209,10 @@ public Flowable streamCall(SpeechSynthesisParam param) { if (result.getAudioFrame() != null) { if (recvAudioLength == 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("[TtsV2] first package delay: " + getFirstPackageDelay() + " ms"); + log.debug( + "[TtsV2] first package delay: {} ms, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } recvAudioLength += (double) result.getAudioFrame().capacity() @@ -193,16 +220,19 @@ public Flowable streamCall(SpeechSynthesisParam param) { long current = System.currentTimeMillis(); double current_rtf = (current - startStreamTimeStamp) / recvAudioLength; log.debug( - "[TtsV2] Recv Audio Binary: " - + result.getAudioFrame().capacity() - + " bytes, total audio " - + recvAudioLength - + " ms, current_rtf: " - + current_rtf); + "[TtsV2] Recv Audio Binary: {} bytes, total audio {} ms, current_rtf: {}, requestId={}", + result.getAudioFrame().capacity(), + recvAudioLength, + current_rtf, + getRequestIdForLog()); try { channel.write(result.getAudioFrame()); } catch (IOException e) { - log.error("Failed to write audio: {}", result.getAudioFrame(), e); + log.error( + "Failed to write audio: {}, requestId={}", + result.getAudioFrame(), + getRequestIdForLog(), + e); } } return result; @@ -214,7 +244,11 @@ public Flowable streamCall(SpeechSynthesisParam param) { channel.close(); outputStream.close(); } catch (IOException e) { - log.error("Failed to close channel: {}", channel, e); + log.error( + "Failed to close channel: {}, requestId={}", + channel, + getRequestIdForLog(), + e); } }) .doOnError( @@ -225,7 +259,11 @@ public Flowable streamCall(SpeechSynthesisParam param) { timestamps.clear(); audioData = null; } catch (IOException ex) { - log.error("Failed to close channel: {}", channel, ex); + log.error( + "Failed to close channel: {}, requestId={}", + channel, + getRequestIdForLog(), + ex); } }); } catch (NoApiKeyException e) { 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 8d866870..a650f50a 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizer.java @@ -55,6 +55,7 @@ public final class SpeechSynthesizer { @Getter private ByteBuffer audioData; private ByteArrayOutputStream outputStream; private String preRequestId = null; + private AtomicReference lastRequestId = new AtomicReference<>(null); private boolean isFirst = true; private AtomicBoolean canceled = new AtomicBoolean(false); private final AtomicReference lastError = new AtomicReference<>(null); @@ -160,6 +161,7 @@ public void updateParamAndCallback( this.textEmitter = null; this.isFirst = true; this.lastError.set(null); + this.lastRequestId.set(null); this.asyncCall = this.callback != null; } @@ -228,6 +230,10 @@ public String getLastRequestId() { return preRequestId; } + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } + /** * Returns the given options, or a default instance routed to the dedicated audio client * (useDefaultClient=false) when null. @@ -252,6 +258,7 @@ public Flowable streamingCallAsFlowable(Flowable startStreamTimeStamp = System.currentTimeMillis(); recvAudioLength = 0; preRequestId = UUID.randomUUID().toString(); + lastRequestId.set(null); Flowable inputStream = textStream.map(text -> new TextStreamItem(text)); return duplexApi .duplexCall( @@ -262,10 +269,16 @@ public Flowable streamingCallAsFlowable(Flowable .filter(item -> !canceled.get()) .doOnNext( result -> { + if (result.getRequestId() != null && !result.getRequestId().isEmpty()) { + lastRequestId.set(result.getRequestId()); + } if (result.getAudioFrame() != null) { if (recvAudioLength == 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("[TtsV2] first package delay: " + getFirstPackageDelay() + " ms"); + log.debug( + "[TtsV2] first package delay: {} ms, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } recvAudioLength += (double) result.getAudioFrame().capacity() @@ -273,12 +286,11 @@ public Flowable streamingCallAsFlowable(Flowable long current = System.currentTimeMillis(); double currentRtf = (current - startStreamTimeStamp) / recvAudioLength; log.debug( - "[TtsV2] Recv Audio Binary: " - + result.getAudioFrame().capacity() - + " bytes, total audio " - + recvAudioLength - + " ms, current_rtf: " - + currentRtf); + "[TtsV2] Recv Audio Binary: {} bytes, total audio {} ms, current_rtf: {}, requestId={}", + result.getAudioFrame().capacity(), + recvAudioLength, + currentRtf, + getRequestIdForLog()); } }); } @@ -294,6 +306,7 @@ public Flowable callAsFlowable(String text) startStreamTimeStamp = System.currentTimeMillis(); recvAudioLength = 0; preRequestId = UUID.randomUUID().toString(); + lastRequestId.set(null); return duplexApi .duplexCall( StreamInputTtsParamWithStream.fromStreamInputTtsParam( @@ -315,10 +328,16 @@ public Flowable callAsFlowable(String text) .map(SpeechSynthesisResult::fromDashScopeResult) .doOnNext( result -> { + if (result.getRequestId() != null && !result.getRequestId().isEmpty()) { + lastRequestId.set(result.getRequestId()); + } if (result.getAudioFrame() != null) { if (recvAudioLength == 0) { firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("[TtsV2] first package delay: " + getFirstPackageDelay() + " ms"); + log.debug( + "[TtsV2] first package delay: {} ms, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } recvAudioLength += (double) result.getAudioFrame().capacity() @@ -326,12 +345,11 @@ public Flowable callAsFlowable(String text) long current = System.currentTimeMillis(); double currentRtf = (current - startStreamTimeStamp) / recvAudioLength; log.debug( - "[TtsV2] Recv Audio Binary: " - + result.getAudioFrame().capacity() - + " bytes, total audio " - + recvAudioLength - + " ms, current_rtf: " - + currentRtf); + "[TtsV2] Recv Audio Binary: {} bytes, total audio {} ms, current_rtf: {}, requestId={}", + result.getAudioFrame().capacity(), + recvAudioLength, + currentRtf, + getRequestIdForLog()); } }); } @@ -371,7 +389,7 @@ private void startStream(boolean enableSsml) { } cmdBuffer.clear(); } - log.debug("set textEmitter"); + log.debug("set textEmitter, requestId={}", getRequestIdForLog()); textEmitter = emitter; } }, @@ -384,6 +402,7 @@ private void startStream(boolean enableSsml) { startLatch = new AtomicReference<>(new CountDownLatch(1)); firstAudioLatch = new AtomicReference<>(new CountDownLatch(1)); preRequestId = UUID.randomUUID().toString(); + lastRequestId.set(null); // Bind the callback that owns this session: once updateParamAndCallback() installs a new // callback, a late event from this session must not be delivered to the next one. final ResultCallback sessionCallback = this.callback; @@ -396,6 +415,9 @@ private void startStream(boolean enableSsml) { @Override public void onEvent(DashScopeResult message) { + if (message.getRequestId() != null && !message.getRequestId().isEmpty()) { + lastRequestId.set(message.getRequestId()); + } if (message.getEvent() == WebSocketEventType.TASK_STARTED.getValue()) { startLatch.get().countDown(); return; @@ -427,7 +449,10 @@ public void onEvent(DashScopeResult message) { if (recvAudioLength == 0) { firstAudioLatch.get().countDown(); firstPackageTimeStamp = System.currentTimeMillis(); - log.debug("[TtsV2] first package delay: " + getFirstPackageDelay() + " ms"); + log.debug( + "[TtsV2] first package delay: {} ms, requestId={}", + getFirstPackageDelay(), + getRequestIdForLog()); } recvAudioLength += (double) speechSynthesisResult.getAudioFrame().capacity() @@ -435,23 +460,30 @@ public void onEvent(DashScopeResult message) { long current = System.currentTimeMillis(); double currentRtf = (current - startStreamTimeStamp) / recvAudioLength; log.debug( - "[TtsV2] Recv Audio Binary: " - + speechSynthesisResult.getAudioFrame().capacity() - + " bytes, total audio " - + recvAudioLength - + " ms, current_rtf: " - + currentRtf); + "[TtsV2] Recv Audio Binary: {} bytes, total audio {} ms, current_rtf: {}, requestId={}", + speechSynthesisResult.getAudioFrame().capacity(), + recvAudioLength, + currentRtf, + getRequestIdForLog()); if (!asyncCall) { try { channel.write(speechSynthesisResult.getAudioFrame()); } catch (IOException e) { log.error( - "Failed to write audio: {}", speechSynthesisResult.getAudioFrame(), e); + "Failed to write audio: {}, requestId={}", + speechSynthesisResult.getAudioFrame(), + getRequestIdForLog(), + e); } } } } catch (Exception e) { - log.error("Failed to parse response: {}", message, e); + log.error( + "Failed to parse response, event: {}, output: {}, requestId={}", + message.getEvent(), + message.getOutput(), + getRequestIdForLog(), + e); sessionCallback.onError(e); } if (speechSynthesisResult.getRequestId() == null) { @@ -462,7 +494,7 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { - log.debug("[TtsV2] onComplete"); + log.debug("[TtsV2] onComplete, requestId={}", getRequestIdForLog()); synchronized (SpeechSynthesizer.this) { state = SpeechSynthesisState.IDLE; } @@ -470,7 +502,7 @@ public void onComplete() { try { outputStream.close(); } catch (IOException e) { - log.error("Failed to close channel: {}", e); + log.error("Failed to close channel, requestId={}", getRequestIdForLog(), e); } sessionCallback.onComplete(); if (stopLatch.get() != null) { @@ -487,7 +519,10 @@ public void onError(Exception e) { // releaseLatches() only unblocks streamingComplete()/streamingCall(); a // callback-driven caller waits on onComplete/onError and would otherwise never // get any signal. Report onComplete, matching the cancel-then-success path. - log.debug("[TtsV2] error after cancel, reported as complete: " + e.getMessage()); + log.debug( + "[TtsV2] error after cancel, reported as complete: {}, requestId={}", + e.getMessage(), + getRequestIdForLog()); synchronized (SpeechSynthesizer.this) { state = SpeechSynthesisState.IDLE; } @@ -562,10 +597,10 @@ private void submitText(String text) { "State invalid: expect stream input tts state is started but " + state.getValue())); } if (textEmitter == null) { - log.debug("submitText to new emitter: " + text); + log.debug("submitText to new emitter: {}, requestId={}", text, getRequestIdForLog()); cmdBuffer.add(AsyncCmdBuffer.builder().item(new TextStreamItem(text)).build()); } else { - log.debug("submitText to emitter: " + text); + log.debug("submitText to emitter: {}, requestId={}", text, getRequestIdForLog()); textEmitter.onNext(new TextStreamItem(text)); } } @@ -580,7 +615,10 @@ private void submitText(String text) { * will wait indefinitely. Throws TimeoutError exception if it times out. */ public void streamingComplete(long completeTimeoutMillis) { - log.debug("streamingComplete with timeout: " + completeTimeoutMillis); + log.debug( + "streamingComplete with timeout: {}, requestId={}", + completeTimeoutMillis, + getRequestIdForLog()); synchronized (this) { if (state != SpeechSynthesisState.TTS_STARTED) { throw new ApiException( @@ -588,10 +626,10 @@ public void streamingComplete(long completeTimeoutMillis) { "State invalid: expect stream input tts state is started but " + state.getValue())); } if (textEmitter == null) { - log.debug("adding stop to new emitter"); + log.debug("adding stop to new emitter, requestId={}", getRequestIdForLog()); cmdBuffer.add(AsyncCmdBuffer.builder().isStop(true).build()); } else { - log.debug("adding stop to emitter"); + log.debug("adding stop to emitter, requestId={}", getRequestIdForLog()); textEmitter.onComplete(); } } @@ -599,17 +637,18 @@ public void streamingComplete(long completeTimeoutMillis) { if (stopLatch.get() != null) { try { if (completeTimeoutMillis > 0) { - log.debug("start waiting for stopLatch"); + log.debug("start waiting for stopLatch, requestId={}", getRequestIdForLog()); if (!stopLatch.get().await(completeTimeoutMillis, TimeUnit.MILLISECONDS)) { throw new RuntimeException("TimeoutError: waiting for streaming complete"); } } else { - log.debug("start waiting for stopLatch"); + log.debug("start waiting for stopLatch, requestId={}", getRequestIdForLog()); stopLatch.get().await(); } - log.debug("stopLatch is done"); + log.debug("stopLatch is done, requestId={}", getRequestIdForLog()); } catch (InterruptedException ignored) { - log.error("Interrupted while waiting for streaming complete"); + log.error( + "Interrupted while waiting for streaming complete, requestId={}", getRequestIdForLog()); } } } @@ -686,11 +725,10 @@ public void streamingCall(String text) { "TimeoutError: waiting for task started more than " + this.startedTimeout + " ms."); } log.debug( - "get started within " - + this.startedTimeout - + " ms, cost: " - + (System.currentTimeMillis() - startTime) - + " ms"); + "get started within {} ms, cost: {} ms, requestId={}", + this.startedTimeout, + System.currentTimeMillis() - startTime, + getRequestIdForLog()); } this.submitText(text); if (this.firstAudioTimeout > 0 && this.firstAudioTimeout - this.startedTimeout > 0) { @@ -706,14 +744,14 @@ public void streamingCall(String text) { + " ms."); } log.debug( - "get first audio within " - + this.firstAudioTimeout - + " ms, cost: " - + (System.currentTimeMillis() - startTime) - + " ms"); + "get first audio within {} ms, cost: {} ms, requestId={}", + this.firstAudioTimeout, + System.currentTimeMillis() - startTime, + getRequestIdForLog()); } } catch (InterruptedException ignored) { - log.error("Interrupted while waiting for streaming complete"); + log.error( + "Interrupted while waiting for streaming complete, requestId={}", getRequestIdForLog()); } } else { this.submitText(text); @@ -732,10 +770,10 @@ public void streamingFlush(JsonObject params) { "State invalid: expect stream input tts state is started but " + state.getValue())); } if (textEmitter == null) { - log.debug("submitFlush to new emitter"); + log.debug("submitFlush to new emitter, requestId={}", getRequestIdForLog()); cmdBuffer.add(AsyncCmdBuffer.builder().item(new TextStreamItem(true, params)).build()); } else { - log.debug("submitFlush to emitter"); + log.debug("submitFlush to emitter, requestId={}", getRequestIdForLog()); textEmitter.onNext(new TextStreamItem(true, params)); } } @@ -781,11 +819,10 @@ public void onError(Exception e) {} "TimeoutError: waiting for task started more than " + this.startedTimeout + " ms."); } log.debug( - "get started within " - + this.startedTimeout - + " ms, cost: " - + (System.currentTimeMillis() - startTime) - + " ms"); + "get started within {} ms, cost: {} ms, requestId={}", + this.startedTimeout, + System.currentTimeMillis() - startTime, + getRequestIdForLog()); } this.submitText(text); if (this.firstAudioTimeout > 0 && this.firstAudioTimeout - this.startedTimeout > 0) { @@ -799,14 +836,14 @@ public void onError(Exception e) {} "TimeoutError: waiting for first audio more than " + this.firstAudioTimeout + " ms."); } log.debug( - "get first audio within " - + this.firstAudioTimeout - + " ms, cost: " - + (System.currentTimeMillis() - startTime) - + " ms"); + "get first audio within {} ms, cost: {} ms, requestId={}", + this.firstAudioTimeout, + System.currentTimeMillis() - startTime, + getRequestIdForLog()); } } catch (InterruptedException ignored) { - log.error("Interrupted while waiting for streaming complete"); + log.error( + "Interrupted while waiting for streaming complete, requestId={}", getRequestIdForLog()); } catch (RuntimeException e) { // The real cause may already have been recorded by the error callback, in which case the // exception raised here only masks it. 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 edd38047..8bf4cd67 100644 --- a/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java +++ b/src/main/java/com/alibaba/dashscope/audio/ttsv2/SpeechSynthesizerV2.java @@ -139,6 +139,20 @@ public String getLastRequestId() { return preRequestId; } + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(taskId); + } + + private String getRequestIdForLog(JsonObject message) { + if (message != null && message.has("header")) { + JsonObject header = message.getAsJsonObject("header"); + if (header.has(TASK_ID) && !header.get(TASK_ID).isJsonNull()) { + return DashScopeHeaders.requestIdOrUnknown(header.get(TASK_ID).getAsString()); + } + } + return getRequestIdForLog(); + } + private void checkConnectStatus() { websocketRequest.checkStatus(); } @@ -165,7 +179,10 @@ public void close() { try { websocketRequest.close(); } catch (Exception e) { - log.warn("Failed to close websocket connection: " + e.getMessage()); + log.warn( + "Failed to close websocket connection: {}, requestId={}", + e.getMessage(), + getRequestIdForLog()); } } releaseLatches(); @@ -209,7 +226,7 @@ private synchronized void sendTaskMessage(String action, JsonObject input) { wsMessage.add("header", JsonUtils.toJsonObject(bailianHeader)); wsMessage.add("payload", JsonUtils.toJsonObject(payload)); - log.debug("sendTaskMessage: {}", wsMessage.toString()); + log.debug("sendTaskMessage: {}, requestId={}", wsMessage.toString(), getRequestIdForLog()); websocketRequest.sendTextMessage(wsMessage.toString()); } @@ -251,7 +268,7 @@ public void stopSynthesizer() { @Override public void onOpen() { - log.info("WebSocket connection opened"); + log.info("WebSocket connection opened, requestId={}", getRequestIdForLog()); if (callback != null) { callback.onOpen(null); } @@ -259,7 +276,7 @@ public void onOpen() { @Override public void onMessage(WebSocket webSocket, String text) { - log.debug("Received text message: " + text); + log.debug("Received text message: {}, requestId={}", text, getRequestIdForLog()); try { JsonObject messageObj = JsonParser.parseString(text).getAsJsonObject(); if (messageObj.has("header")) { @@ -281,19 +298,24 @@ public void onMessage(WebSocket webSocket, String text) { handleResultGenerated(messageObj); break; default: - log.warn("Unknown event: " + event); + log.warn("Unknown event: {}, requestId={}", event, getRequestIdForLog(messageObj)); break; } } } } catch (Exception e) { - log.error("Error processing text message: " + e.getMessage(), e); + log.error( + "Error processing text message: {}, requestId={}", + e.getMessage(), + getRequestIdForLog(), + e); } } @Override public void onMessage(WebSocket webSocket, ByteBuffer bytes) { - log.debug("Received binary message, size: {}", bytes.remaining()); + log.debug( + "Received binary message, size: {}, requestId={}", bytes.remaining(), getRequestIdForLog()); try { ByteBuffer audioFrame = ByteBuffer.allocate(bytes.remaining()); audioFrame.put(bytes); @@ -312,7 +334,7 @@ public void onMessage(WebSocket webSocket, ByteBuffer bytes) { recvAudioLength += bytes.remaining(); } catch (Exception e) { - log.error("Error processing binary message", e); + log.error("Error processing binary message, requestId={}", getRequestIdForLog(), e); if (callback != null) { callback.onError(e); } @@ -347,12 +369,13 @@ public void onError(WebSocket webSocket, Throwable t) { @Override public void onClose(int code, String reason) { - log.warn("WebSocket connection closed: " + reason + " (" + code + ")"); + log.warn( + "WebSocket connection closed: {} ({}), requestId={}", reason, code, getRequestIdForLog()); releaseLatches(); } private void handleTaskStarted(JsonObject message) { - log.info("Task started"); + log.info("Task started, requestId={}", getRequestIdForLog(message)); state = SpeechSynthesisState.TTS_STARTED; firstPackageTimeStamp = -1; CountDownLatch startLatch = this.startLatch.get(); @@ -362,7 +385,7 @@ private void handleTaskStarted(JsonObject message) { } private void handleTaskFinished(JsonObject message) { - log.info("Task finished"); + log.info("Task finished, requestId={}", getRequestIdForLog(message)); CountDownLatch stopLatch = this.stopLatch.get(); if (stopLatch != null) { stopLatch.countDown(); @@ -378,7 +401,7 @@ private void handleTaskFinished(JsonObject message) { } private void handleTaskFailed(JsonObject message) { - log.error("Task failed: " + message.toString()); + log.error("Task failed: {}, requestId={}", message, getRequestIdForLog(message)); if (callback != null) { String errorMessage = "Unknown error"; if (message.has("header") && message.getAsJsonObject("header").has("error_message")) { @@ -400,7 +423,7 @@ private void handleTaskFailed(JsonObject message) { } private void handleResultGenerated(JsonObject message) { - log.debug("Result generated: " + message.toString()); + log.debug("Result generated: {}, requestId={}", message, getRequestIdForLog(message)); if (callback == null) { return; } @@ -475,7 +498,10 @@ private void startStream() throws NoApiKeyException, InterruptedException { } public void streamingComplete(long completeTimeoutMillis) { - log.debug("streamingComplete with timeout: " + completeTimeoutMillis); + log.debug( + "streamingComplete with timeout: {}, requestId={}", + completeTimeoutMillis, + getRequestIdForLog()); synchronized (this) { if (state != SpeechSynthesisState.TTS_STARTED) { throw new ApiException( @@ -489,17 +515,18 @@ public void streamingComplete(long completeTimeoutMillis) { try { if (completeTimeoutMillis > 0) { - log.debug("start waiting for stopLatch"); + log.debug("start waiting for stopLatch, requestId={}", getRequestIdForLog()); if (!stopLatch.await(completeTimeoutMillis, TimeUnit.MILLISECONDS)) { throw new RuntimeException("TimeoutError: waiting for streaming complete"); } } else { - log.debug("start waiting for stopLatch"); + log.debug("start waiting for stopLatch, requestId={}", getRequestIdForLog()); stopLatch.await(); } - log.debug("stopLatch is done"); + log.debug("stopLatch is done, requestId={}", getRequestIdForLog()); } catch (InterruptedException ignored) { - log.error("Interrupted while waiting for streaming complete"); + log.error( + "Interrupted while waiting for streaming complete, requestId={}", getRequestIdForLog()); Thread.currentThread().interrupt(); } } @@ -537,7 +564,10 @@ public void streamingCall(String text) { this.submitText(text); } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 恢复中断状态 - log.error("Interrupted while waiting for streaming complete", e); + log.error( + "Interrupted while waiting for streaming complete, requestId={}", + getRequestIdForLog(), + e); throw new ApiException(e); } catch (NoApiKeyException e) { throw new ApiException(e); @@ -578,7 +608,10 @@ public ByteBuffer call(String text, long timeoutMillis) throws RuntimeException this.submitText(text); } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 恢复中断状态 - log.error("Interrupted while waiting for streaming complete", e); + log.error( + "Interrupted while waiting for streaming complete, requestId={}", + getRequestIdForLog(), + e); throw new ApiException(e); } catch (NoApiKeyException e) { throw new ApiException(e); diff --git a/src/main/java/com/alibaba/dashscope/common/DashScopeResult.java b/src/main/java/com/alibaba/dashscope/common/DashScopeResult.java index c7eab7ab..09026d68 100644 --- a/src/main/java/com/alibaba/dashscope/common/DashScopeResult.java +++ b/src/main/java/com/alibaba/dashscope/common/DashScopeResult.java @@ -3,6 +3,7 @@ package com.alibaba.dashscope.common; import com.alibaba.dashscope.exception.ApiException; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.HalfDuplexRequest; import com.alibaba.dashscope.protocol.NetworkResponse; import com.alibaba.dashscope.protocol.Protocol; @@ -35,13 +36,15 @@ public Boolean isBinaryOutput() { protected T fromResponse(Protocol protocol, NetworkResponse response) throws ApiException { this.setHeaders(changeHeaders(response.getHeaders())); + fillRequestIdFromHeaders(); if (protocol == Protocol.WEBSOCKET) { if (response.getBinary() == null) { String message = response.getMessage(); if (message == null || message.isEmpty()) { log.warn( - "WebSocket response message is null or empty, httpStatusCode: {}", - response.getHttpStatusCode()); + "WebSocket response message is null or empty, httpStatusCode: {}, requestId={}", + response.getHttpStatusCode(), + getRequestIdOrUnknown()); setInternalError(); return (T) this; } @@ -61,7 +64,7 @@ protected T fromResponse(Protocol protocol, NetworkResponse r this.setMessage(""); return (T) this; } - JsonObject jsonObject = parseJson(message, "Failed to parse HTTP response as JSON: {}"); + JsonObject jsonObject = parseJson(message, "Failed to parse HTTP response as JSON"); if (jsonObject == null) { return (T) this; } @@ -83,6 +86,7 @@ public T fromResponse( return fromResponse(protocol, response); } this.setHeaders(changeHeaders(response.getHeaders())); + fillRequestIdFromHeaders(); if (protocol == Protocol.WEBSOCKET) { if (response.getBinary() == null) { this.output = parseJson(response.getMessage(), "Failed to parse WebSocket message"); @@ -112,6 +116,7 @@ public T fromResponse( Protocol protocol, NetworkResponse response, boolean isFlattenResult, HalfDuplexRequest req) throws ApiException { this.setHeaders(changeHeaders(response.getHeaders())); + fillRequestIdFromHeaders(); if ((response.getHeaders().containsKey("X-DashScope-OutputEncrypted".toLowerCase()) || req.isEncryptRequest()) && protocol == Protocol.HTTP) { @@ -234,7 +239,7 @@ private JsonObject parseJson(String raw, String logMsg) { try { return JsonUtils.parse(raw); } catch (Exception e) { - log.error(logMsg, e); + log.error("{}, requestId={}", logMsg, getRequestIdOrUnknown(), e); setInternalError(); return null; } @@ -263,4 +268,14 @@ private Map changeHeaders(Map> headers) { (v1, v2) -> v1, java.util.LinkedHashMap::new)); } + + private void fillRequestIdFromHeaders() { + if (getRequestId() != null && !getRequestId().isEmpty()) { + return; + } + String requestId = DashScopeHeaders.findRequestId(getHeaders()); + if (requestId != null) { + setRequestId(requestId); + } + } } diff --git a/src/main/java/com/alibaba/dashscope/common/Result.java b/src/main/java/com/alibaba/dashscope/common/Result.java index 8193eabf..6c69372e 100644 --- a/src/main/java/com/alibaba/dashscope/common/Result.java +++ b/src/main/java/com/alibaba/dashscope/common/Result.java @@ -35,6 +35,10 @@ public abstract class Result { /** The message from server response */ private String message; + public String getRequestIdOrUnknown() { + return requestId == null || requestId.isEmpty() ? "unknown" : requestId; + } + /** * Load data from the server output. * diff --git a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java index 2e119db8..70c4cbe8 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/MultiModalDialog.java @@ -8,6 +8,7 @@ import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import com.alibaba.dashscope.utils.JsonUtils; @@ -57,6 +58,12 @@ public class MultiModalDialog { private String currentDialogId = ""; // Current dialogue ID + private final AtomicReference lastRequestId = new AtomicReference<>(null); + + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } + @SuperBuilder private static class AsyncCmdBuffer { // Asynchronous command buffer class @Builder.Default private boolean isStop = false; // Stop flag, defaults to false @@ -214,6 +221,7 @@ public void start() { BackpressureStrategy.BUFFER); closed.set(false); // Reset idempotent close guard for reuse across sessions + lastRequestId.set(null); stopLatch.set(new CountDownLatch(1)); // Initializes stop signal latch String preTaskId = @@ -228,13 +236,20 @@ public void start() { new ResultCallback() { // Registers callback handler @Override public void onEvent(DashScopeResult message) { + if (message.getRequestId() != null && !message.getRequestId().isEmpty()) { + lastRequestId.set(message.getRequestId()); + } if (message.isBinaryOutput()) { callback.onSpeechAudioData( (ByteBuffer) message.getOutput()); // Handles speech output data return; } - log.info("onEvent: {}", message); // Logs event information + log.info( + "onEvent: event={}, output={}, requestId={}", + message.getEvent(), + message.getOutput(), + message.getRequestIdOrUnknown()); // Logs event information if (message.getOutput() != null) { JsonObject output = @@ -311,8 +326,9 @@ public void onEvent(DashScopeResult message) { if (message.getEvent().equals("task-started")) { callback.onConnected(); log.debug( - "MultiModalDialog connected, state is {}", - currentState.getValue()); // Logs connection status + "MultiModalDialog connected, state is {}, requestId={}", + currentState.getValue(), + message.getRequestIdOrUnknown()); // Logs connection status } } } @@ -486,7 +502,9 @@ public void stop() { // Use timeout to prevent deadlock: if server doesn't respond in 30s, force close boolean completed = latch.await(30, TimeUnit.SECONDS); if (!completed) { - log.warn("stop() timed out waiting for server acknowledgement, forcing close"); + log.warn( + "stop() timed out waiting for server acknowledgement, forcing close, requestId={}", + getRequestIdForLog()); close(); } } catch (InterruptedException e) { @@ -555,8 +573,9 @@ public void sendAudioData(ByteBuffer audioFrame) { synchronized (MultiModalDialog.this) { // Synchronized block ensures thread safety if (conversationEmitter != null) { log.debug( - "submitAudioFrame to new emitter: " - + audioFrame); // Logs submission of audio frame information + "submitAudioFrame to new emitter: {}, requestId={}", + audioFrame, + getRequestIdForLog()); // Logs submission of audio frame information conversationEmitter.onNext(audioFrame); // Sends audio frame data } else { DialogBuffer.add( @@ -580,8 +599,9 @@ private void sendTextFrame( synchronized (MultiModalDialog.this) { // Synchronized block ensures thread safety if (conversationEmitter != null) { log.debug( - "submitText to new emitter: {}", - textFrame); // Logs submission of text frame information + "submitText to new emitter: {}, requestId={}", + textFrame, + getRequestIdForLog()); // Logs submission of text frame information if ("RequestToRespond".equals(textFrame) || "UpdateInfo".equals(textFrame)) { // RequestToRespond and UpdateInfo have special inputs and parameters diff --git a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java index 552edc24..3d60f3ca 100644 --- a/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java +++ b/src/main/java/com/alibaba/dashscope/multimodal/tingwu/TingWuRealtime.java @@ -9,6 +9,7 @@ import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.protocol.ApiServiceOption; import com.alibaba.dashscope.protocol.ConnectionOptions; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.protocol.Protocol; import com.alibaba.dashscope.protocol.StreamingMode; import io.reactivex.BackpressureStrategy; @@ -49,6 +50,11 @@ private static class AsyncCmdBuffer { private long onCompleteTimeStamp = -1; private String preRequestId = null; private boolean isListenState = false; + private final AtomicReference lastRequestId = new AtomicReference<>(null); + + private String getRequestIdForLog() { + return DashScopeHeaders.requestIdOrUnknown(lastRequestId.get()); + } @SuperBuilder private static class TingWuRealtimeWithStream extends TingWuRealtimeParam { @@ -185,7 +191,10 @@ public void call(TingWuRealtimeParam param, TingWuRealtimeCallback callback) { new ResultCallback() { @Override public void onEvent(DashScopeResult message) { - log.debug("Response Result :" + message); + if (lastRequestId.get() == null && message.getRequestId() != null) { + lastRequestId.set(message.getRequestId()); + } + log.debug("Response Result: {}, requestId={}", message, getRequestIdForLog()); TingWuRealtimeResult tingWuRealtimeResult = TingWuRealtimeResult.fromDashScopeResult(message); @@ -221,7 +230,10 @@ public void onEvent(DashScopeResult message) { @Override public void onComplete() { onCompleteTimeStamp = System.currentTimeMillis(); - log.debug("last package delay: " + getLastPackageDelay()); + log.debug( + "last package delay: {}, requestId={}", + getLastPackageDelay(), + getRequestIdForLog()); callback.onClosed(); if (stopLatch.get() != null) { stopLatch.get().countDown(); @@ -246,7 +258,7 @@ public void onError(Exception e) { stopLatch.get().countDown(); } } - log.debug("TingWu Realtime client started"); + log.debug("TingWu Realtime client started, requestId={}", getRequestIdForLog()); } /** @@ -262,7 +274,7 @@ public void sendAudioFrame(ByteBuffer audioFrame) { if (this.startStreamTimeStamp < 0) { this.startStreamTimeStamp = System.currentTimeMillis(); } - log.debug("send audio frame: " + audioFrame.remaining()); + log.debug("send audio frame: {}, requestId={}", audioFrame.remaining(), getRequestIdForLog()); synchronized (this) { if (audioEmitter == null || !isListenState) { cmdBuffer.add(AsyncCmdBuffer.builder().audioFrame(audioFrame).build()); @@ -305,6 +317,7 @@ private void reset() { this.firstPackageTimeStamp = -1; this.stopStreamTimeStamp = -1; this.onCompleteTimeStamp = -1; + this.lastRequestId.set(null); preRequestId = null; } diff --git a/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingResult.java b/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingResult.java index c7c9427b..50949bee 100644 --- a/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingResult.java +++ b/src/main/java/com/alibaba/dashscope/nlp/understanding/UnderstandingResult.java @@ -30,7 +30,10 @@ public static UnderstandingResult fromDashScopeResult(DashScopeResult dashScopeR JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), UnderstandingOutput.class)); } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/protocol/DashScopeHeaders.java b/src/main/java/com/alibaba/dashscope/protocol/DashScopeHeaders.java index fc9b1ba7..f4b69b69 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/DashScopeHeaders.java +++ b/src/main/java/com/alibaba/dashscope/protocol/DashScopeHeaders.java @@ -14,6 +14,36 @@ public final class DashScopeHeaders { private static final String SDK_CLIENT = "java-sdk"; private static final String SDK_SESSION_ID = UUID.randomUUID().toString(); + public static String findRequestId(Map headers) { + if (headers == null) { + return null; + } + for (Map.Entry entry : headers.entrySet()) { + String name = entry.getKey(); + if (name == null + || !("x-request-id".equalsIgnoreCase(name) + || "x-dashscope-requestid".equalsIgnoreCase(name) + || "x-dashscope-request-id".equalsIgnoreCase(name))) { + continue; + } + Object value = entry.getValue(); + if (value instanceof Iterable) { + for (Object item : (Iterable) value) { + if (item != null && !item.toString().isEmpty()) { + return item.toString(); + } + } + } else if (value != null && !value.toString().isEmpty()) { + return value.toString(); + } + } + return null; + } + + public static String requestIdOrUnknown(String requestId) { + return requestId == null || requestId.isEmpty() ? "unknown" : requestId; + } + public static String userAgent() { return userAgent(null); } diff --git a/src/main/java/com/alibaba/dashscope/protocol/okhttp/LoggingInterceptor.java b/src/main/java/com/alibaba/dashscope/protocol/okhttp/LoggingInterceptor.java index 9b164f23..68359955 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/okhttp/LoggingInterceptor.java +++ b/src/main/java/com/alibaba/dashscope/protocol/okhttp/LoggingInterceptor.java @@ -2,6 +2,7 @@ package com.alibaba.dashscope.protocol.okhttp; +import com.alibaba.dashscope.protocol.DashScopeHeaders; import com.alibaba.dashscope.utils.StringUtils; import java.io.IOException; import lombok.extern.slf4j.Slf4j; @@ -18,15 +19,24 @@ public Response intercept(Interceptor.Chain chain) throws IOException { long t1 = System.nanoTime(); log.info( StringUtils.format( - "Sending request %s on %s%n%s", request.url(), chain.connection(), request.headers())); + "Sending request %s on %s%n%s, requestId=%s", + request.url(), + chain.connection(), + request.headers(), + DashScopeHeaders.requestIdOrUnknown( + DashScopeHeaders.findRequestId(request.headers().toMultimap())))); Response response = chain.proceed(request); long t2 = System.nanoTime(); log.info( StringUtils.format( - "Received response for %s in %.1fms%n%s", - response.request().url(), (t2 - t1) / 1e6d, response.headers())); + "Received response for %s in %.1fms%n%s, requestId=%s", + response.request().url(), + (t2 - t1) / 1e6d, + response.headers(), + DashScopeHeaders.requestIdOrUnknown( + DashScopeHeaders.findRequestId(response.headers().toMultimap())))); return response; } diff --git a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpHttpClient.java b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpHttpClient.java index 40fb6d2e..f8749825 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpHttpClient.java +++ b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpHttpClient.java @@ -9,6 +9,7 @@ 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.HalfDuplexClient; import com.alibaba.dashscope.protocol.HalfDuplexRequest; import com.alibaba.dashscope.protocol.HttpMethod; @@ -53,6 +54,14 @@ public final class OkHttpHttpClient implements HalfDuplexClient { private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json; charset=utf-8"); + private static String requestId(Response response) { + if (response == null) { + return "unknown"; + } + return DashScopeHeaders.requestIdOrUnknown( + DashScopeHeaders.findRequestId(response.headers().toMultimap())); + } + private Status parseStreamEventData(String data) { try { JsonObject jsonResponse = JsonUtils.parse(data); @@ -368,7 +377,10 @@ private void handleSSEEvent( Response response, HalfDuplexRequest req, AtomicBoolean terminated) { - log.debug(StringUtils.format("Event: id %s, type: %s, data: %s", id, eventType, data)); + log.debug( + StringUtils.format( + "Event: id %s, type: %s, data: %s, requestId=%s", + id, eventType, data, requestId(response))); if (terminated.get()) { return; } @@ -393,7 +405,10 @@ private void handleSSEEvent( isFlattenResult, req)); } else if (SSEEventType.DONE.equals(eventType)) { // event done ignore message - log.debug(StringUtils.format("Ignore event id: %s, type: %s, data: %s", id, eventType, data)); + log.debug( + StringUtils.format( + "Ignore event id: %s, type: %s, data: %s, requestId=%s", + id, eventType, data, requestId(response))); } else if (eventType != null) { // process assistant events. emitter.onNext( @@ -530,7 +545,10 @@ public void onEvent( java.lang.String id, java.lang.String type, java.lang.String data) { - log.debug(StringUtils.format("Event: id %s, type: %s, data: %s", id, type, data)); + log.debug( + StringUtils.format( + "Event: id %s, type: %s, data: %s, requestId=%s", + id, type, data, requestId(response))); if (terminated.get()) { return; } diff --git a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClient.java b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClient.java index cc0045b4..d909f429 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClient.java +++ b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClient.java @@ -63,6 +63,7 @@ public class OkHttpWebSocketClient extends WebSocketListener private static final Pattern CLIENT_ERROR_STATUS_LINE = Pattern.compile("was '4\\d\\d "); private OkHttpClient client; private WebSocket webSocketClient; + private volatile String requestId; // indicate the websocket connection is established. private AtomicBoolean isOpen = new AtomicBoolean(false); protected AtomicBoolean isClosed = new AtomicBoolean(false); @@ -87,6 +88,37 @@ public OkHttpWebSocketClient(OkHttpClient client, boolean passTaskStarted) { this.passTaskStarted.set(passTaskStarted); } + protected String getRequestIdOrUnknown() { + return DashScopeHeaders.requestIdOrUnknown(requestId); + } + + private void updateRequestId(String requestId) { + if (requestId != null && !requestId.isEmpty()) { + this.requestId = requestId; + } + } + + private void setRequestIdFromStartMessage(JsonObject startMessage) { + this.requestId = null; + try { + if (startMessage == null || !startMessage.has("header")) { + return; + } + JsonObject header = startMessage.getAsJsonObject("header"); + if (header != null && header.has("task_id") && !header.get("task_id").isJsonNull()) { + this.requestId = header.get("task_id").getAsString(); + } + } catch (RuntimeException ignored) { + // Request processing keeps its original validation and error handling. + } + } + + private void updateRequestIdFromResponse(Response response) { + if (response != null && (requestId == null || requestId.isEmpty())) { + updateRequestId(DashScopeHeaders.findRequestId(response.headers().toMultimap())); + } + } + private Request buildConnectionRequest( String apiKey, boolean isSecurityCheck, @@ -169,7 +201,7 @@ private void establishWebSocketClient( while (true) { // Bail out immediately if cancel() has been called if (isClosed.get()) { - log.debug("Connection cancelled, stop reconnecting."); + log.debug("Connection cancelled, stop reconnecting. requestId={}", getRequestIdOrUnknown()); return; } try { @@ -202,7 +234,7 @@ private void establishWebSocketClient( } catch (Throwable ex) { attempts += 1; errorMessage = ex.getMessage() != null ? ex.getMessage() : ex.toString(); - log.error(errorMessage); + log.error("{}, requestId={}", errorMessage, getRequestIdOrUnknown()); if (errorMessage.contains(Constants.NO_API_KEY_ERROR)) { throw ex; } @@ -212,9 +244,10 @@ private void establishWebSocketClient( // so give up immediately. if (isClientError(httpStatusCode, errorMessage)) { log.warn( - "Websocket handshake refused with http status {}, will not retry: {}", + "Websocket handshake refused with http status {}, will not retry: {}, requestId={}", httpStatusCode, - errorMessage); + errorMessage, + getRequestIdOrUnknown()); break; } if (attempts >= MAX_CONNECTION_TIMES) { @@ -223,7 +256,9 @@ private void establishWebSocketClient( } // Check again before sleeping if (isClosed.get()) { - log.debug("Connection cancelled during retry, stop reconnecting."); + log.debug( + "Connection cancelled during retry, stop reconnecting. requestId={}", + getRequestIdOrUnknown()); return; } if (!sleepBeforeRetry(attempts)) { @@ -335,7 +370,9 @@ public void onClosed(WebSocket webSocket, int code, String reason) { // calls to this // listener will be made. log.debug( - StringUtils.format("WebSocket %s closed: %d, %s", webSocket.toString(), code, reason)); + StringUtils.format( + "WebSocket %s closed: %d, %s, requestId=%s", + webSocket.toString(), code, reason, getRequestIdOrUnknown())); isOpen.set(false); } @@ -359,7 +396,10 @@ public void onClosing(WebSocket webSocket, int code, String reason) { // endpoint that understands only text data MAY send this if it receives a // binary message) webSocket.close(code, null); - log.debug(StringUtils.format("Websocket is closing, code: %s, reasion: %s", code, reason)); + log.debug( + StringUtils.format( + "Websocket is closing, code: %s, reasion: %s, requestId=%s", + code, reason, getRequestIdOrUnknown())); if (responseEmitter != null && !responseEmitter.isCancelled()) { responseEmitter.onComplete(); } else { // close on idle, such as server close the connection. @@ -369,13 +409,16 @@ public void onClosing(WebSocket webSocket, int code, String reason) { @Override public void onFailure(WebSocket webSocket, Throwable t, Response response) { + updateRequestIdFromResponse(response); // Invoked when a web socket has been closed due to an error reading from or // writing to the network. // Both outgoing and incoming messages may have been lost. No further calls to // this listener will be made. if (isClosed.get()) { - log.debug("called close before but not working, close again in onFailure."); + log.debug( + "called close before but not working, close again in onFailure. requestId={}", + getRequestIdOrUnknown()); close(1001, "call closed before"); // Do not leave establishWebSocketClient() waiting for the 60s timeout. releaseConnectionWaiter(); @@ -392,7 +435,7 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) { try { responseBody = body.string(); } catch (IOException ex) { - log.error(ex.getMessage()); + log.error("{}, requestId={}", ex.getMessage(), getRequestIdOrUnknown()); } } } @@ -404,7 +447,7 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) { // Appended, not inlined, to keep the message prefix stable for log based diagnostics. failureMessage = failureMessage + ", http status: " + httpStatusCode; } - log.error(failureMessage); + log.error("{}, requestId={}", failureMessage, getRequestIdOrUnknown()); isOpen.set(false); if (connectionEmitter != null && !connectionEmitter.isCancelled()) { connectionEmitter.onError(new WebSocketConnectException(failureMessage, httpStatusCode, t)); @@ -412,26 +455,37 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) { // error on request responseEmitter.onError(new Exception(failureMessage, t)); } else { - log.error(failureMessage); + log.error("{}, requestId={}", failureMessage, getRequestIdOrUnknown()); } } @Override public void onMessage(WebSocket webSocket, String text) { if (isClosed.get()) { - log.debug("called close before but not working, close again in onMessage."); + log.debug( + "called close before but not working, close again in onMessage. requestId={}", + getRequestIdOrUnknown()); close(1001, "call closed before"); return; } - log.debug(text); - // Invoked when a text (type 0x1) message has been received. - if (!isFirstMessage.get()) { - log.debug("Receive first package."); + boolean firstMessage = !isFirstMessage.get(); + if (firstMessage) { isFirstMessage.set(true); } + boolean messageLogged = false; try { // Check different message. WebSocketResponse response = JsonUtils.fromJson(text, WebSocketResponse.class); + if (response.header != null) { + updateRequestId(response.header.taskId); + } + log.debug("{}, requestId={}", text, getRequestIdOrUnknown()); + messageLogged = true; + // Invoked when a text (type 0x1) message has been received. + if (firstMessage) { + log.debug("Receive first package. requestId={}", getRequestIdOrUnknown()); + firstMessage = false; + } switch (response.header.event) { case TASK_STARTED: // if has payload, call onNext. @@ -454,7 +508,9 @@ public void onMessage(WebSocket webSocket, String text) { } break; case TASK_FAILED: - log.error(StringUtils.format("Receive task_failed message: %s", text)); + log.error( + StringUtils.format( + "Receive task_failed message: %s, requestId=%s", text, getRequestIdOrUnknown())); Status st = Status.builder() .code(response.header.code) @@ -467,7 +523,10 @@ public void onMessage(WebSocket webSocket, String text) { if (!responseEmitter.isCancelled()) { responseEmitter.onError(new ApiException(st)); } else { - log.error(StringUtils.format("Something wrong, receive task failed message: %s", text)); + log.error( + StringUtils.format( + "Something wrong, receive task failed message: %s, requestId=%s", + text, getRequestIdOrUnknown())); } case TASK_FINISHED: // check the payload and usage is null. @@ -503,6 +562,12 @@ public void onMessage(WebSocket webSocket, String text) { .build())); } } catch (Throwable ex) { + if (!messageLogged) { + log.debug("{}, requestId={}", text, getRequestIdOrUnknown()); + } + if (firstMessage) { + log.debug("Receive first package. requestId={}", getRequestIdOrUnknown()); + } responseEmitter.onError( new ApiException( Status.builder() @@ -518,12 +583,14 @@ public void onMessage(WebSocket webSocket, String text) { public void onMessage(WebSocket webSocket, ByteString bytes) { // Invoked when a binary (type 0x2) message has been received. if (isClosed.get()) { - log.debug("called close before but not working, close again in onMessage."); + log.debug( + "called close before but not working, close again in onMessage. requestId={}", + getRequestIdOrUnknown()); close(1001, "call closed before"); return; } if (!isFirstMessage.get()) { - log.debug("Receive first binary package."); + log.debug("Receive first binary package. requestId={}", getRequestIdOrUnknown()); isFirstMessage.set(true); } responseEmitter.onNext( @@ -536,13 +603,16 @@ public void onMessage(WebSocket webSocket, ByteString bytes) { @Override public void onOpen(WebSocket webSocket, Response response) { + updateRequestIdFromResponse(response); // the connection has been accepted by the remote peer and may begin // transmitting messages // Invoked when a web socket has been accepted by the remote peer and may begin // transmitting // messages.. if (isClosed.get()) { - log.debug("called close before but not working, close again in onOpen."); + log.debug( + "called close before but not working, close again in onOpen. requestId={}", + getRequestIdOrUnknown()); close(1001, "call closed before"); // Do not leave establishWebSocketClient() waiting for the 60s timeout. releaseConnectionWaiter(); @@ -564,7 +634,9 @@ protected void sendTextWithRetry( String module) { // Guard: skip if already cancelled if (isClosed.get()) { - log.debug("sendTextWithRetry skipped: connection already closed."); + log.debug( + "sendTextWithRetry skipped: connection already closed. requestId={}", + getRequestIdOrUnknown()); return; } // simple retry with fixed delay, no strategy @@ -579,13 +651,15 @@ protected void sendTextWithRetry( if (passTaskStarted.get()) { // when pass througn task started, no need to retry. if (webSocketClient == null) { - log.warn("webSocketClient is null, cannot send message."); + log.warn( + "webSocketClient is null, cannot send message. requestId={}", getRequestIdOrUnknown()); return; } - log.info("Sending message: " + message); + log.info("Sending message: {}, requestId={}", message, getRequestIdOrUnknown()); Boolean isOk = webSocketClient.send(message); if (!isOk) { - log.warn("Send request failed, return without retry."); + log.warn( + "Send request failed, return without retry. requestId={}", getRequestIdOrUnknown()); } return; } @@ -595,10 +669,11 @@ protected void sendTextWithRetry( return; } if (webSocketClient == null) { - log.warn("webSocketClient is null, cannot send message."); + log.warn( + "webSocketClient is null, cannot send message. requestId={}", getRequestIdOrUnknown()); return; } - log.debug("Sending message: " + message); + log.debug("Sending message: {}, requestId={}", message, getRequestIdOrUnknown()); Boolean isOk = webSocketClient.send(message); if (isOk) { break; @@ -607,7 +682,8 @@ protected void sendTextWithRetry( apiKey, isSecurityCheck, workspace, customHeaders, baseWebSocketUrl, module); log.warn( StringUtils.format( - "Send request failed, the connection may closed, will reconnect and send again")); + "Send request failed, the connection may closed, will reconnect and send again, requestId=%s", + getRequestIdOrUnknown())); } Observable.timer(5000, TimeUnit.MILLISECONDS).blockingSingle(); ++retryCount; @@ -640,7 +716,9 @@ protected void sendBinaryWithRetry( return; } if (webSocketClient == null) { - log.warn("webSocketClient is null, cannot send binary message."); + log.warn( + "webSocketClient is null, cannot send binary message. requestId={}", + getRequestIdOrUnknown()); return; } Boolean isOk = webSocketClient.send(message); @@ -651,7 +729,8 @@ protected void sendBinaryWithRetry( apiKey, isSecurityCheck, workspace, customHeaders, baseWebSocketUrl, module); log.warn( StringUtils.format( - "Send request failed, the connection may closed, will reconnect and send again")); + "Send request failed, the connection may closed, will reconnect and send again, requestId=%s", + getRequestIdOrUnknown())); } Observable.timer(5000, TimeUnit.MILLISECONDS).blockingSingle(); ++retryCount; @@ -659,12 +738,14 @@ protected void sendBinaryWithRetry( } private void sendBatchRequest(HalfDuplexRequest req) { + JsonObject startMessage = req.getStartTaskMessage(); + setRequestIdFromStartMessage(startMessage); if (req.getWebsocketBinaryData() != null) { // send start-task. sendTextWithRetry( req.getApiKey(), req.isSecurityCheck(), - JsonUtils.toJson(req.getStartTaskMessage()), + JsonUtils.toJson(startMessage), req.getWorkspace(), req.getHeaders(), req.getBaseWebSocketUrl(), @@ -683,7 +764,7 @@ private void sendBatchRequest(HalfDuplexRequest req) { sendTextWithRetry( req.getApiKey(), req.isSecurityCheck(), - JsonUtils.toJson(req.getStartTaskMessage()), + JsonUtils.toJson(startMessage), req.getWorkspace(), req.getHeaders(), req.getBaseWebSocketUrl(), @@ -831,6 +912,7 @@ protected void executeStreamRequest(FullDuplexRequest req) { pendingResponseError.set(null); JsonObject startMessage = req.getStartTaskMessage(); + setRequestIdFromStartMessage(startMessage); onBeforeSendStartMessage(startMessage); String taskId = startMessage.get("header").getAsJsonObject().get("task_id").getAsString(); // send start message out. @@ -888,18 +970,25 @@ protected void executeStreamRequest(FullDuplexRequest req) { req.getModule()); } } catch (Throwable ex) { - log.error(StringUtils.format("sendStreamData exception: %s", ex.getMessage())); + log.error( + StringUtils.format( + "sendStreamData exception: %s, requestId=%s", + ex.getMessage(), getRequestIdOrUnknown())); emitResponseError(ex); } }, err -> { - log.error(StringUtils.format("Get stream data error!")); + log.error( + StringUtils.format( + "Get stream data error! requestId=%s", getRequestIdOrUnknown())); emitResponseError(err); }, new Action() { @Override public void run() throws Exception { - log.debug(StringUtils.format("Stream data send completed!")); + log.debug( + StringUtils.format( + "Stream data send completed! requestId=%s", getRequestIdOrUnknown())); sendTextWithRetry( req.getApiKey(), req.isSecurityCheck(), @@ -918,7 +1007,10 @@ public void run() throws Exception { d.dispose(); } } catch (Throwable ex) { - log.error(StringUtils.format("sendStreamData exception: %s", ex.getMessage())); + log.error( + StringUtils.format( + "sendStreamData exception: %s, requestId=%s", + ex.getMessage(), getRequestIdOrUnknown())); emitResponseError(ex); } } @@ -937,9 +1029,15 @@ private void joinSendFuture(CompletableFuture future) { future.join(); } } catch (CancellationException ex) { - log.error("Sending streaming data cancelled", ex.getMessage()); + log.error( + "Sending streaming data cancelled: {}, requestId={}", + ex.getMessage(), + getRequestIdOrUnknown()); } catch (CompletionException ex) { - log.error("Sending streaming data exception", ex.getMessage()); + log.error( + "Sending streaming data exception: {}, requestId={}", + ex.getMessage(), + getRequestIdOrUnknown()); emitResponseError(ex.getCause() != null ? ex.getCause() : ex); } } diff --git a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClientForAudio.java b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClientForAudio.java index 7917cb3e..3d2614ba 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClientForAudio.java +++ b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpWebSocketClientForAudio.java @@ -89,7 +89,10 @@ public OkHttpWebSocketClientForAudio(OkHttpClient client, boolean passTaskStarte @Override protected void onBeforeSendStartMessage(JsonObject startMessage) { - log.info("send run-task request {}", JsonUtils.toJson(startMessage)); + log.info( + "send run-task request {}, requestId={}", + JsonUtils.toJson(startMessage), + getRequestIdOrUnknown()); } @Override diff --git a/src/main/java/com/alibaba/dashscope/rerank/TextReRankResult.java b/src/main/java/com/alibaba/dashscope/rerank/TextReRankResult.java index 734f8014..d7c3119a 100644 --- a/src/main/java/com/alibaba/dashscope/rerank/TextReRankResult.java +++ b/src/main/java/com/alibaba/dashscope/rerank/TextReRankResult.java @@ -43,7 +43,10 @@ public static TextReRankResult fromDashScopeResult(DashScopeResult dashScopeResu JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), TextReRankOutput.class)); } else { - log.error("Result no output: {}", dashScopeResult); + log.error( + "Result no output: {}, requestId={}", + dashScopeResult, + dashScopeResult.getRequestIdOrUnknown()); } return result; } diff --git a/src/main/java/com/alibaba/dashscope/task/AsyncTaskListResult.java b/src/main/java/com/alibaba/dashscope/task/AsyncTaskListResult.java index 5087111a..da40fb5b 100644 --- a/src/main/java/com/alibaba/dashscope/task/AsyncTaskListResult.java +++ b/src/main/java/com/alibaba/dashscope/task/AsyncTaskListResult.java @@ -46,7 +46,10 @@ public static AsyncTaskListResult fromDashScopeResult(DashScopeResult dashScopeR rs.message = dashScopeResult.getMessage(); return rs; } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return null; } diff --git a/src/main/java/com/alibaba/dashscope/tokenizers/TokenizationResult.java b/src/main/java/com/alibaba/dashscope/tokenizers/TokenizationResult.java index 828995f0..2347442b 100644 --- a/src/main/java/com/alibaba/dashscope/tokenizers/TokenizationResult.java +++ b/src/main/java/com/alibaba/dashscope/tokenizers/TokenizationResult.java @@ -29,7 +29,10 @@ public static TokenizationResult fromDashScopeResult(DashScopeResult dashScopeRe JsonUtils.fromJsonObject( (JsonObject) dashScopeResult.getOutput(), TokenizationOutput.class); } else { - log.error(StringUtils.format("Result no output: %s", dashScopeResult)); + log.error( + StringUtils.format( + "Result no output: %s, requestId=%s", + dashScopeResult, dashScopeResult.getRequestIdOrUnknown())); } return result; } diff --git a/src/test/java/com/alibaba/dashscope/TestDashScopeResultEmptyBody.java b/src/test/java/com/alibaba/dashscope/TestDashScopeResultEmptyBody.java index f69392da..81071461 100644 --- a/src/test/java/com/alibaba/dashscope/TestDashScopeResultEmptyBody.java +++ b/src/test/java/com/alibaba/dashscope/TestDashScopeResultEmptyBody.java @@ -14,6 +14,9 @@ import com.alibaba.dashscope.protocol.NetworkResponse; import com.alibaba.dashscope.protocol.Protocol; import com.google.gson.JsonObject; +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; import java.util.Collections; import org.junit.jupiter.api.Test; @@ -136,4 +139,43 @@ public void testMalformedBodyIsStillAnInternalError() throws Exception { assertEquals(PublicErrorDef.INTERNAL_ERROR.getStatusCode(), result.getStatusCode()); assertEquals(PublicErrorDef.INTERNAL_ERROR.getErrorCode(), result.getCode()); } + + @Test + public void testBodyRequestIdOverridesResponseHeader() throws Exception { + NetworkResponse response = + NetworkResponse.builder() + .headers( + Collections.singletonMap( + "x-request-id", Collections.singletonList("request-from-header"))) + .message("{\"request_id\":\"request-from-body\",\"output\":{}}") + .httpStatusCode(200) + .build(); + DashScopeResult result = new DashScopeResult().fromResponse(Protocol.HTTP, response, false); + assertEquals("request-from-body", result.getRequestId()); + } + + @Test + public void testMalformedBodyLogsHeaderRequestId() throws Exception { + NetworkResponse response = + NetworkResponse.builder() + .headers( + Collections.singletonMap( + "x-request-id", Collections.singletonList("request-from-header"))) + .message("not json at all") + .httpStatusCode(200) + .build(); + PrintStream originalErr = System.err; + ByteArrayOutputStream output = new ByteArrayOutputStream(); + DashScopeResult result; + try { + System.setErr(new PrintStream(output, true, StandardCharsets.UTF_8.name())); + result = new DashScopeResult().fromResponse(Protocol.HTTP, response, false); + } finally { + System.setErr(originalErr); + } + assertEquals("request-from-header", result.getRequestId()); + assertTrue( + new String(output.toByteArray(), StandardCharsets.UTF_8) + .contains("requestId=request-from-header")); + } } diff --git a/src/test/java/com/alibaba/dashscope/TestMultiModalConversation.java b/src/test/java/com/alibaba/dashscope/TestMultiModalConversation.java index 02ebe136..63134227 100644 --- a/src/test/java/com/alibaba/dashscope/TestMultiModalConversation.java +++ b/src/test/java/com/alibaba/dashscope/TestMultiModalConversation.java @@ -3,6 +3,8 @@ package com.alibaba.dashscope; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import com.alibaba.dashscope.aigc.multimodalconversation.MultiModalConversation; import com.alibaba.dashscope.aigc.multimodalconversation.MultiModalConversationMessage; @@ -10,6 +12,7 @@ import com.alibaba.dashscope.aigc.multimodalconversation.MultiModalConversationResult; import com.alibaba.dashscope.aigc.multimodalconversation.MultiModalMessageItemImage; import com.alibaba.dashscope.aigc.multimodalconversation.MultiModalMessageItemText; +import com.alibaba.dashscope.common.DashScopeResult; import com.alibaba.dashscope.common.MultiModalMessage; import com.alibaba.dashscope.common.Role; import com.alibaba.dashscope.exception.ApiException; @@ -17,8 +20,13 @@ import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.exception.UploadFileException; import com.alibaba.dashscope.utils.Constants; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.PrintStream; +import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.List; import lombok.extern.slf4j.Slf4j; import okhttp3.MediaType; import okhttp3.mockwebserver.MockResponse; @@ -30,6 +38,8 @@ import org.junit.jupiter.api.parallel.Execution; import org.junit.jupiter.api.parallel.ExecutionMode; import org.junitpioneer.jupiter.SetEnvironmentVariable; +import org.slf4j.LoggerFactory; +import org.slf4j.simple.SimpleLogger; @Execution(ExecutionMode.SAME_THREAD) @Slf4j @@ -39,6 +49,35 @@ public class TestMultiModalConversation { MediaType.parse("application/json; charset=utf-8"); MockWebServer server; + private static class SdkLogCapture implements AutoCloseable { + private final PrintStream originalErr = System.err; + private final ByteArrayOutputStream output = new ByteArrayOutputStream(); + private final SimpleLogger logger; + private final Field levelField; + private final int originalLevel; + + SdkLogCapture(Class loggerClass, boolean debug) throws Exception { + logger = (SimpleLogger) LoggerFactory.getLogger(loggerClass); + levelField = SimpleLogger.class.getDeclaredField("currentLogLevel"); + levelField.setAccessible(true); + originalLevel = levelField.getInt(logger); + if (debug) { + levelField.setInt(logger, 10); + } + System.setErr(new PrintStream(output, true, StandardCharsets.UTF_8.name())); + } + + String output() { + return new String(output.toByteArray(), StandardCharsets.UTF_8); + } + + @Override + public void close() throws Exception { + System.setErr(originalErr); + levelField.setInt(logger, originalLevel); + } + } + @BeforeEach public void before() { this.server = new MockWebServer(); @@ -97,4 +136,58 @@ public void testSendAndReceive() assertEquals(request.getMethod(), "POST"); assertEquals(request.getPath(), "/services/aigc/multimodal-generation/generation"); } + + @Test + public void testStreamIgnoresUsageOnlyFrameAndUsesHeaderRequestId() throws Exception { + server.enqueue( + new MockResponse() + .setHeader("content-type", "text/event-stream") + .setHeader("x-request-id", "request-from-header") + .setBody( + "data: {\"usage\":{\"input_tokens\":26,\"output_tokens\":0}}\n\n" + + "data: {\"output\":{\"choices\":[{\"finish_reason\":\"stop\"," + + "\"message\":{\"role\":\"assistant\"," + + "\"content\":[{\"text\":\"正常结果\"}]}}]}," + + "\"usage\":{\"input_tokens\":26,\"output_tokens\":4}}\n\n" + + "data: [DONE]\n\n")); + Constants.baseHttpApiUrl = String.format("http://127.0.0.1:%s", server.getPort()); + MultiModalConversationParam param = + MultiModalConversationParam.builder() + .model("qwen3.7-plus") + .message( + MultiModalConversationMessage.builder() + .role(Role.USER.getValue()) + .content(Arrays.asList(new MultiModalMessageItemText("检查材料"))) + .build()) + .build(); + + List results; + try (SdkLogCapture capture = new SdkLogCapture(MultiModalConversation.class, true)) { + results = new MultiModalConversation().streamCall(param).toList().blockingGet(); + assertTrue(capture.output().contains("requestId=request-from-header")); + } + + assertEquals(1, results.size()); + MultiModalConversationResult result = results.get(0); + assertEquals("request-from-header", result.getRequestId()); + assertNotNull(result.getOutput()); + assertEquals( + "正常结果", + result.getOutput().getChoices().get(0).getMessage().getContent().get(0).get("text")); + } + + @Test + public void testResultNoOutputLogsRequestId() throws Exception { + DashScopeResult withRequestId = new DashScopeResult(); + withRequestId.setRequestId("request-from-body"); + try (SdkLogCapture capture = new SdkLogCapture(MultiModalConversationResult.class, false)) { + MultiModalConversationResult.fromDashScopeResult(withRequestId); + assertTrue(capture.output().contains("requestId=request-from-body")); + } + + try (SdkLogCapture capture = new SdkLogCapture(MultiModalConversationResult.class, false)) { + MultiModalConversationResult.fromDashScopeResult(new DashScopeResult()); + assertTrue(capture.output().contains("requestId=unknown")); + } + } } diff --git a/src/test/java/com/alibaba/dashscope/protocol/TestDashScopeHeaders.java b/src/test/java/com/alibaba/dashscope/protocol/TestDashScopeHeaders.java index 7018c5b4..a6b4e674 100644 --- a/src/test/java/com/alibaba/dashscope/protocol/TestDashScopeHeaders.java +++ b/src/test/java/com/alibaba/dashscope/protocol/TestDashScopeHeaders.java @@ -14,6 +14,7 @@ import com.alibaba.dashscope.exception.ApiException; import com.alibaba.dashscope.exception.NoApiKeyException; import com.alibaba.dashscope.task.AsyncTaskParam; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.UUID; @@ -212,4 +213,24 @@ public void testAsyncTaskPollRequestCarriesTasksModule() throws NoApiKeyExceptio assertEquals("tasks", request.getModule()); assertEquals(CLIENT_PREFIX + "/tasks", request.getHttpRequest().getHeaders().get(SDK_CLIENT)); } + + @Test + public void testFindRequestIdSupportsResponseHeaderVariants() { + assertEquals( + "request-1", + DashScopeHeaders.findRequestId( + Collections.singletonMap("X-Request-Id", Collections.singletonList("request-1")))); + assertEquals( + "request-2", + DashScopeHeaders.findRequestId( + Collections.singletonMap("x-dashscope-requestid", "request-2"))); + assertEquals( + "request-3", + DashScopeHeaders.findRequestId( + Collections.singletonMap("X-DashScope-Request-Id", "request-3"))); + assertNull(DashScopeHeaders.findRequestId(null)); + assertEquals("unknown", DashScopeHeaders.requestIdOrUnknown(null)); + assertEquals("unknown", DashScopeHeaders.requestIdOrUnknown("")); + assertEquals("request-4", DashScopeHeaders.requestIdOrUnknown("request-4")); + } } diff --git a/src/test/java/com/alibaba/dashscope/protocol/okhttp/TestWebSocketRequestIdLogging.java b/src/test/java/com/alibaba/dashscope/protocol/okhttp/TestWebSocketRequestIdLogging.java new file mode 100644 index 00000000..98e00266 --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/protocol/okhttp/TestWebSocketRequestIdLogging.java @@ -0,0 +1,153 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. + +package com.alibaba.dashscope.protocol.okhttp; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.alibaba.dashscope.audio.protocol.AudioWebsocketCallback; +import com.alibaba.dashscope.audio.protocol.AudioWebsocketRequest; +import com.alibaba.dashscope.common.DashScopeResult; +import io.reactivex.BackpressureStrategy; +import io.reactivex.Flowable; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.PrintStream; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.WebSocket; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.slf4j.LoggerFactory; +import org.slf4j.simple.SimpleLogger; + +@Execution(ExecutionMode.SAME_THREAD) +public class TestWebSocketRequestIdLogging { + private static class SdkLogCapture implements AutoCloseable { + private final PrintStream originalErr = System.err; + private final ByteArrayOutputStream output = new ByteArrayOutputStream(); + private final SimpleLogger logger; + private final Field levelField; + private final int originalLevel; + + SdkLogCapture(Class loggerClass) throws Exception { + logger = (SimpleLogger) LoggerFactory.getLogger(loggerClass); + levelField = SimpleLogger.class.getDeclaredField("currentLogLevel"); + levelField.setAccessible(true); + originalLevel = levelField.getInt(logger); + levelField.setInt(logger, 10); + System.setErr(new PrintStream(output, true, StandardCharsets.UTF_8.name())); + } + + String output() { + return new String(output.toByteArray(), StandardCharsets.UTF_8); + } + + @Override + public void close() throws Exception { + System.setErr(originalErr); + levelField.setInt(logger, originalLevel); + } + } + + private static OkHttpWebSocketClient newClient() { + OkHttpWebSocketClient client = new OkHttpWebSocketClient(new OkHttpClient(), false); + Flowable.create( + emitter -> client.responseEmitter = emitter, BackpressureStrategy.BUFFER) + .subscribe(); + return client; + } + + private static Response failureResponse(String requestId) { + return new Response.Builder() + .request(new Request.Builder().url("http://localhost/").build()) + .protocol(okhttp3.Protocol.HTTP_1_1) + .code(500) + .message("failed") + .header("x-request-id", requestId) + .build(); + } + + private static AudioWebsocketRequest newAudioRequest() throws Exception { + AudioWebsocketRequest request = new AudioWebsocketRequest(); + Field callbackField = AudioWebsocketRequest.class.getDeclaredField("callback"); + callbackField.setAccessible(true); + callbackField.set( + request, + new AudioWebsocketCallback() { + @Override + public void onOpen() {} + + @Override + public void onMessage(WebSocket webSocket, String text) {} + + @Override + public void onMessage(WebSocket webSocket, ByteBuffer buffer) {} + + @Override + public void onError(WebSocket webSocket, Throwable t) {} + + @Override + public void onClose(int code, String reason) {} + }); + return request; + } + + @Test + public void testOkHttpLogsUnknownBeforeRequestIdIsAvailable() throws Exception { + OkHttpWebSocketClient client = new OkHttpWebSocketClient(new OkHttpClient(), false); + try (SdkLogCapture capture = new SdkLogCapture(OkHttpWebSocketClient.class)) { + client.onFailure(null, new IOException("failed"), null); + assertTrue(capture.output().contains("requestId=unknown")); + } + } + + @Test + public void testOkHttpLogsTaskIdFromMessage() throws Exception { + try (SdkLogCapture capture = new SdkLogCapture(OkHttpWebSocketClient.class)) { + newClient() + .onMessage( + null, + "{\"header\":{\"task_id\":\"websocket-task\",\"event\":\"task-started\"}," + + "\"payload\":{}}"); + String output = capture.output(); + assertTrue(output.contains("requestId=websocket-task")); + assertTrue(output.contains("Receive first package. requestId=websocket-task")); + } + } + + @Test + public void testOkHttpHandshakeFailureLogsResponseHeaderRequestId() throws Exception { + OkHttpWebSocketClient client = new OkHttpWebSocketClient(new OkHttpClient(), false); + try (SdkLogCapture capture = new SdkLogCapture(OkHttpWebSocketClient.class)) { + client.onFailure(null, new IOException("handshake failed"), failureResponse("handshake-id")); + assertTrue(capture.output().contains("requestId=handshake-id")); + } + } + + @Test + public void testAudioLogsUnknownBeforeRequestIdIsAvailable() throws Exception { + AudioWebsocketRequest request = newAudioRequest(); + try (SdkLogCapture capture = new SdkLogCapture(AudioWebsocketRequest.class)) { + request.onClosed(null, 1000, "bye"); + assertTrue(capture.output().contains("requestId=unknown")); + } + } + + @Test + public void testAudioTaskIdIsNotOverwrittenByFailureResponseHeader() throws Exception { + AudioWebsocketRequest request = newAudioRequest(); + try (SdkLogCapture capture = new SdkLogCapture(AudioWebsocketRequest.class)) { + request.onMessage(null, "{\"header\":{\"task_id\":\"audio-task\"}}"); + request.onFailure(null, new IOException("failed"), failureResponse("handshake-id")); + String output = capture.output(); + assertTrue(output.contains("requestId=audio-task")); + assertFalse(output.contains("requestId=handshake-id")); + } + } +}