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 d4ebe62..40fb6d2 100644 --- a/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpHttpClient.java +++ b/src/main/java/com/alibaba/dashscope/protocol/okhttp/OkHttpHttpClient.java @@ -131,6 +131,21 @@ private Status parseFailedJson(int statusCode, String body) { } } + private Status tryParseErrorJson(int statusCode, String body) { + try { + JsonObject jsonResponse = JsonUtils.parse(body); + if (jsonResponse == null + || (!jsonResponse.has(ApiKeywords.CODE) + && !jsonResponse.has(ApiKeywords.MESSAGE) + && !jsonResponse.has(ApiKeywords.ERROR))) { + return null; + } + return parseFailedJson(statusCode, body); + } catch (Throwable e) { + return null; + } + } + private Status parseFailed(Response response, Throwable th) { if (response == null) { String message = th == null ? "Get response failed!" : th.getMessage(); @@ -158,30 +173,55 @@ private Status parseFailed(Response response, Throwable th) { } return parseFailedJson(response.code(), body); } else if (contentType != null && contentType.toLowerCase().contains("text/event-stream")) { + String body = null; try { - String body = response.body().string(); + body = response.body().string(); + } catch (Throwable e) { + // The body may be partially consumed or the connection already broken. + } + if (body != null) { for (String part : body.split("\n")) { part = part.trim(); if (part.startsWith("data:")) { - body = part.replace("data:", ""); - return parseFailedJson(response.code(), body); + return parseFailedJson(response.code(), part.substring("data:".length()).trim()); } } - return Status.builder() - .statusCode(response.code()) - .code(ErrorType.RESPONSE_ERROR.getValue()) - .message(body) - .isJson(false) - .build(); - } catch (IOException e) { - return Status.builder() - .statusCode(response.code()) - .code(ErrorType.RESPONSE_ERROR.getValue()) - .message("Failed read response body: " + e.getMessage()) - .isJson(true) - .build(); + // The gateway may write a bare JSON error without SSE framing. + Status jsonStatus = tryParseErrorJson(response.code(), body); + if (jsonStatus != null) { + return jsonStatus; + } } + String message; + if (body != null && !body.isEmpty()) { + message = body; + } else if (th != null && th.getMessage() != null) { + message = th.getMessage(); + } else { + message = response.message(); + } + return Status.builder() + .statusCode(response.code()) + .code(ErrorType.RESPONSE_ERROR.getValue()) + .message(message) + .isJson(false) + .build(); } else { + // The error body may still be JSON even if the content type says otherwise. + String body = null; + try { + if (response.body() != null) { + body = response.peekBody(16 * 1024).string(); + } + } catch (Throwable e) { + // Ignore unreadable bodies and fall back to the HTTP status message. + } + if (body != null && !body.isEmpty()) { + Status jsonStatus = tryParseErrorJson(response.code(), body); + if (jsonStatus != null) { + return jsonStatus; + } + } return Status.builder() .statusCode(response.code()) .code(ErrorType.RESPONSE_ERROR.getValue()) @@ -326,11 +366,19 @@ private void handleSSEEvent( String data, boolean isFlattenResult, Response response, - HalfDuplexRequest req) { + HalfDuplexRequest req, + AtomicBoolean terminated) { log.debug(StringUtils.format("Event: id %s, type: %s, data: %s", id, eventType, data)); + if (terminated.get()) { + return; + } if (SSEEventType.ERROR.equals(eventType)) { Status st = parseStreamEventData(data); - emitter.onError(new ApiException(st)); + // Only one terminal signal may be delivered; a broken connection can + // trigger onFailure right after an error event. + if (terminated.compareAndSet(false, true) && !emitter.isCancelled()) { + emitter.onError(new ApiException(st)); + } } else if (SSEEventType.DATA.equals(eventType) || SSEEventType.RESULT.equals(eventType)) { emitter.onNext( new DashScopeResult() @@ -401,7 +449,14 @@ public void onEvent( java.lang.String type, java.lang.String data) { handleSSEEvent( - emitter, id, type, data, req.getIsFlatten(), response, req); + emitter, + id, + type, + data, + req.getIsFlatten(), + response, + req, + terminated); } @java.lang.Override @@ -417,18 +472,20 @@ public void onFailure( java.lang.Throwable t, Response response) { this.response = response; - terminated.set(true); activeEventSources.remove(eventSource); super.onFailure(eventSource, t, response); - emitter.onError(new ApiException(parseFailed(response, t), t)); + if (terminated.compareAndSet(false, true) && !emitter.isCancelled()) { + emitter.onError(new ApiException(parseFailed(response, t), t)); + } } @java.lang.Override public void onClosed(@NotNull EventSource eventSource) { - terminated.set(true); activeEventSources.remove(eventSource); super.onClosed(eventSource); - emitter.onComplete(); + if (terminated.compareAndSet(false, true)) { + emitter.onComplete(); + } } }); if (terminated.get()) { @@ -460,6 +517,7 @@ private class SSEEventType { public void streamOut(HalfDuplexRequest req, ResultCallback callback) throws NoApiKeyException, ApiException { Request request = buildRequest(req.getHttpRequest()); + final AtomicBoolean terminated = new AtomicBoolean(false); EventSources.createFactory(client) .newEventSource( request, @@ -473,9 +531,14 @@ public void onEvent( java.lang.String type, java.lang.String data) { log.debug(StringUtils.format("Event: id %s, type: %s, data: %s", id, type, data)); + if (terminated.get()) { + return; + } if (SSEEventType.ERROR.equals(type)) { Status st = parseStreamEventData(data); - callback.onError(new ApiException(st)); + if (terminated.compareAndSet(false, true)) { + callback.onError(new ApiException(st)); + } } else if (SSEEventType.DATA.equals(type) || SSEEventType.RESULT.equals(type)) { callback.onEvent( new DashScopeResult() @@ -527,12 +590,16 @@ public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) public void onFailure( @NotNull EventSource eventSource, java.lang.Throwable t, Response response) { this.response = response; - callback.onError(new ApiException(parseFailed(response, t), t)); + if (terminated.compareAndSet(false, true)) { + callback.onError(new ApiException(parseFailed(response, t), t)); + } } @java.lang.Override public void onClosed(EventSource eventSource) { - callback.onComplete(); + if (terminated.compareAndSet(false, true)) { + callback.onComplete(); + } } }); } diff --git a/src/test/java/com/alibaba/dashscope/TestSseErrorHandling.java b/src/test/java/com/alibaba/dashscope/TestSseErrorHandling.java new file mode 100644 index 0000000..ed07c72 --- /dev/null +++ b/src/test/java/com/alibaba/dashscope/TestSseErrorHandling.java @@ -0,0 +1,229 @@ +// Copyright (c) Alibaba, Inc. and its affiliates. + +package com.alibaba.dashscope; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.alibaba.dashscope.api.SynchronizeHalfDuplexApi; +import com.alibaba.dashscope.common.DashScopeResult; +import com.alibaba.dashscope.common.OutputMode; +import com.alibaba.dashscope.common.ResultCallback; +import com.alibaba.dashscope.exception.ApiException; +import com.alibaba.dashscope.exception.NoApiKeyException; +import com.alibaba.dashscope.protocol.ApiServiceOption; +import com.alibaba.dashscope.protocol.HttpMethod; +import com.alibaba.dashscope.protocol.Protocol; +import com.alibaba.dashscope.protocol.StreamingMode; +import com.alibaba.dashscope.utils.Constants; +import io.reactivex.Flowable; +import io.reactivex.plugins.RxJavaPlugins; +import java.io.IOException; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.SocketPolicy; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.junitpioneer.jupiter.SetEnvironmentVariable; + +@Execution(ExecutionMode.SAME_THREAD) +@SetEnvironmentVariable(key = "DASHSCOPE_API_KEY", value = "1234") +public class TestSseErrorHandling { + private static final String ERROR_JSON = + "{\"code\":\"InvalidParameter\",\"message\":\"url error, please check url" + + "\uFF01\",\"request_id\":\"ffbb90bf-1b08-9bfc-9cb3-94eeee5a9258\"}"; + + private SynchronizeHalfDuplexApi syncApi; + private final List undeliverableErrors = new CopyOnWriteArrayList<>(); + + @BeforeEach + public void before() { + ApiServiceOption serviceOption = + ApiServiceOption.builder() + .protocol(Protocol.HTTP) + .httpMethod(HttpMethod.POST) + .streamingMode(StreamingMode.OUT) + .outputMode(OutputMode.ACCUMULATE) + .taskGroup("group") + .task("task") + .function("function") + .build(); + serviceOption.setIsSSE(true); + syncApi = new SynchronizeHalfDuplexApi<>(serviceOption); + undeliverableErrors.clear(); + RxJavaPlugins.setErrorHandler(undeliverableErrors::add); + } + + @AfterEach + public void after() { + RxJavaPlugins.setErrorHandler(null); + } + + private ApiException streamError(MockResponse response) throws IOException, NoApiKeyException { + MockWebServer server = new MockWebServer(); + server.enqueue(response); + Constants.baseHttpApiUrl = String.format("http://127.0.0.1:%s", server.getPort()); + HalfDuplexTestParam param = + HalfDuplexTestParam.builder().model("pre-gateway-mock-async-test").build(); + Flowable flowable = syncApi.streamCall(param); + ApiException e = assertThrows(ApiException.class, () -> flowable.toList().blockingGet()); + server.close(); + return e; + } + + private static MockResponse sseErrorEvent() { + return new MockResponse() + .setResponseCode(200) + .setHeader("content-type", "text/event-stream") + .setBody("event:error\ndata:" + ERROR_JSON + "\n\n"); + } + + private static String sseErrorEventWithPadding() { + StringBuilder sb = new StringBuilder("event:error\ndata:" + ERROR_JSON + "\n\n"); + for (int i = 0; i < 4096; i++) { + sb.append(": padding comment line\n"); + } + return sb.toString(); + } + + @Test + public void testSseErrorEventSurfacesBackendError() throws IOException, NoApiKeyException { + ApiException e = streamError(sseErrorEvent()); + assertEquals("InvalidParameter", e.getStatus().getCode()); + assertTrue(e.getStatus().getMessage().contains("url error")); + assertEquals("ffbb90bf-1b08-9bfc-9cb3-94eeee5a9258", e.getStatus().getRequestId()); + } + + @Test + public void testBrokenStreamAfterErrorEventDeliversSingleError() + throws IOException, InterruptedException, NoApiKeyException { + ApiException e = + streamError( + new MockResponse() + .setResponseCode(200) + .setHeader("content-type", "text/event-stream") + .setBody(sseErrorEventWithPadding()) + .setSocketPolicy(SocketPolicy.DISCONNECT_DURING_RESPONSE_BODY)); + // The backend error from the error event wins over the later connection failure. + assertEquals("InvalidParameter", e.getStatus().getCode()); + assertTrue(e.getStatus().getMessage().contains("url error")); + Thread.sleep(500); + assertTrue( + undeliverableErrors.isEmpty(), + "expected no undeliverable errors, got: " + undeliverableErrors); + } + + @Test + public void testJsonErrorWithNonJsonContentType() throws IOException, NoApiKeyException { + ApiException e = + streamError( + new MockResponse() + .setResponseCode(200) + .setHeader("content-type", "text/plain") + .setBody(ERROR_JSON)); + assertEquals("InvalidParameter", e.getStatus().getCode()); + assertTrue(e.getStatus().getMessage().contains("url error")); + assertEquals("ffbb90bf-1b08-9bfc-9cb3-94eeee5a9258", e.getStatus().getRequestId()); + } + + @Test + public void testRawJsonBodyWithEventStreamContentTypeOnHttpError() + throws IOException, NoApiKeyException { + ApiException e = + streamError( + new MockResponse() + .setResponseCode(400) + .setHeader("content-type", "text/event-stream") + .setBody(ERROR_JSON)); + assertEquals("InvalidParameter", e.getStatus().getCode()); + assertTrue(e.getStatus().getMessage().contains("url error")); + assertEquals("ffbb90bf-1b08-9bfc-9cb3-94eeee5a9258", e.getStatus().getRequestId()); + } + + @Test + public void testBrokenStreamReportsIoMessage() throws IOException, NoApiKeyException { + ApiException e = + streamError( + new MockResponse() + .setResponseCode(200) + .setHeader("content-type", "text/event-stream") + .setBody("data:{\"output\":{\"text\":\"hi\"}}\n\n") + .setSocketPolicy(SocketPolicy.DISCONNECT_DURING_RESPONSE_BODY)); + assertEquals("response_error", e.getStatus().getCode()); + assertTrue( + e.getStatus().getMessage() != null && !e.getStatus().getMessage().isEmpty(), + "expected a non-empty message, got: " + e.getStatus().getMessage()); + } + + @Test + public void testNormalStreamCompletion() + throws IOException, InterruptedException, NoApiKeyException { + MockWebServer server = new MockWebServer(); + server.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("content-type", "text/event-stream") + .setBody( + "data:{\"output\":{\"text\":\"hello\"}}\n\ndata:{\"output\":{\"text\":\"world\"}}\n\n")); + Constants.baseHttpApiUrl = String.format("http://127.0.0.1:%s", server.getPort()); + HalfDuplexTestParam param = HalfDuplexTestParam.builder().model("qwen-turbo").build(); + List results = syncApi.streamCall(param).toList().blockingGet(); + assertEquals(2, results.size()); + Thread.sleep(200); + assertTrue(undeliverableErrors.isEmpty()); + server.close(); + } + + @Test + public void testCallbackReceivesSingleOnError() + throws IOException, InterruptedException, NoApiKeyException { + MockWebServer server = new MockWebServer(); + server.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("content-type", "text/event-stream") + .setBody(sseErrorEventWithPadding()) + .setSocketPolicy(SocketPolicy.DISCONNECT_DURING_RESPONSE_BODY)); + Constants.baseHttpApiUrl = String.format("http://127.0.0.1:%s", server.getPort()); + HalfDuplexTestParam param = + HalfDuplexTestParam.builder().model("pre-gateway-mock-async-test").build(); + AtomicInteger errorCount = new AtomicInteger(0); + AtomicInteger completeCount = new AtomicInteger(0); + Semaphore done = new Semaphore(0); + List errors = new CopyOnWriteArrayList<>(); + syncApi.streamCall( + param, + new ResultCallback() { + @Override + public void onEvent(DashScopeResult message) {} + + @Override + public void onComplete() { + completeCount.incrementAndGet(); + done.release(); + } + + @Override + public void onError(Exception e) { + errorCount.incrementAndGet(); + errors.add(e); + done.release(); + } + }); + assertTrue(done.tryAcquire(10, TimeUnit.SECONDS), "callback not invoked in time"); + Thread.sleep(500); + assertEquals(1, errorCount.get(), "onError should be invoked exactly once"); + assertEquals(0, completeCount.get(), "onComplete should not be invoked after onError"); + assertEquals("InvalidParameter", ((ApiException) errors.get(0)).getStatus().getCode()); + server.close(); + } +}