Skip to content

fix(transport): resolve SSE stream timeout in JdkHttpTransport #1322

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jujn wants to merge 4 commits into
base: main
Choose a base branch
from
Open

fix(transport): resolve SSE stream timeout in JdkHttpTransport #1322

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
260f59b
fix(transport): resolve SSE stream timeout in JdkHttpTransport
jujn Apr 30, 2026
027ae27
fix: review
jujn May 24, 2026
5c0c10e
fix: the issue of responseTimeout being doubled
jujn May 25, 2026
6d73fa2
fix: improve code comments
jujn May 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 56 additions & 1 deletion agentscope-core/src/main/java/io/agentscope/core/model/transport/HttpTransportConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,21 @@ public class HttpTransportConfig {
/** Default connect timeout: 30 seconds. */
public static final Duration DEFAULT_CONNECT_TIMEOUT = Duration.ofSeconds(30);

/** Default read timeout: 5 minutes (for long-running model calls). */
/** Default read timeout: 5 minutes (Overall timeout for non-streaming calls). */
public static final Duration DEFAULT_READ_TIMEOUT = Duration.ofMinutes(5);

/** Default response timeout (TTFT): 5 minutes (Time To First Token for streaming). */
public static final Duration DEFAULT_RESPONSE_TIMEOUT = Duration.ofMinutes(5);

/** Default stream idle timeout: 5 minutes (maximum wait between consecutive data chunks). */
public static final Duration DEFAULT_STREAM_IDLE_TIMEOUT = DEFAULT_READ_TIMEOUT;

/** Default write timeout: 30 seconds. */
public static final Duration DEFAULT_WRITE_TIMEOUT = Duration.ofSeconds(30);

private final Duration connectTimeout;
private final Duration responseTimeout;
private final Duration streamIdleTimeout;
private final Duration readTimeout;
private final Duration writeTimeout;
private final int maxIdleConnections;
Expand All @@ -45,6 +53,8 @@ public class HttpTransportConfig {

private HttpTransportConfig(Builder builder) {
this.connectTimeout = builder.connectTimeout;
this.responseTimeout = builder.responseTimeout;
this.streamIdleTimeout = builder.streamIdleTimeout;
this.readTimeout = builder.readTimeout;
this.writeTimeout = builder.writeTimeout;
this.maxIdleConnections = builder.maxIdleConnections;
Expand All @@ -63,9 +73,30 @@ public Duration getConnectTimeout() {
return connectTimeout;
}

/**
* Get the response timeout (Time To First Token for streaming).
*
* @return the response timeout duration
*/
public Duration getResponseTimeout() {
return responseTimeout;
}

/**
* Get the stream idle timeout (maximum time between two consecutive data chunks).
*
* @return the stream idle timeout duration
*/
public Duration getStreamIdleTimeout() {
return streamIdleTimeout;
}

/**
* Get the read timeout.
*
* <p>For {@link JdkHttpTransport} streaming requests, response and idle timeouts are controlled
* by {@link #getResponseTimeout()} and {@link #getStreamIdleTimeout()} instead.
*
* @return the read timeout duration
*/
public Duration getReadTimeout() {
Expand Down Expand Up @@ -153,6 +184,8 @@ public static HttpTransportConfig defaults() {
*/
public static class Builder {
private Duration connectTimeout = DEFAULT_CONNECT_TIMEOUT;
private Duration responseTimeout = DEFAULT_RESPONSE_TIMEOUT;
private Duration streamIdleTimeout = DEFAULT_STREAM_IDLE_TIMEOUT;
private Duration readTimeout = DEFAULT_READ_TIMEOUT;
private Duration writeTimeout = DEFAULT_WRITE_TIMEOUT;
private int maxIdleConnections = 5;
Expand All @@ -172,6 +205,28 @@ public Builder connectTimeout(Duration connectTimeout) {
return this;
}

/**
* Set the response timeout (Time To First Byte).
*
* @param responseTimeout the response timeout duration
* @return this builder
*/
public Builder responseTimeout(Duration responseTimeout) {
this.responseTimeout = responseTimeout;
return this;
Comment thread
jujn marked this conversation as resolved.
}

/**
* Set the stream idle timeout.
*
* @param streamIdleTimeout the stream idle timeout duration
* @return this builder
*/
public Builder streamIdleTimeout(Duration streamIdleTimeout) {
this.streamIdleTimeout = streamIdleTimeout;
return this;
}

/**
* Set the read timeout.
*
Expand Down
188 changes: 133 additions & 55 deletions agentscope-core/src/main/java/io/agentscope/core/model/transport/JdkHttpTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,13 +34,15 @@
import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.security.cert.X509Certificate;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
Expand Down Expand Up @@ -174,7 +176,7 @@ public HttpResponse execute(HttpRequest request) throws HttpTransportException {
throw new HttpTransportException("Transport has been closed");
}

var jdkRequest = buildJdkRequest(request);
var jdkRequest = buildJdkRequest(request, false);

try {
var response = client.send(jdkRequest, BodyHandlers.ofString());
Expand All @@ -193,55 +195,94 @@ public Flux<String> stream(HttpRequest request) {
return Flux.error(new HttpTransportException("Transport has been closed"));
}

var jdkRequest = buildJdkRequest(request);

// Check status code and read error body immediately when CompletableFuture completes
// to avoid stream being closed before we can read it
CompletableFuture<java.net.http.HttpResponse<InputStream>> future =
client.sendAsync(jdkRequest, BodyHandlers.ofInputStream())
.thenApply(
response -> {
int statusCode = response.statusCode();
if (statusCode < 200 || statusCode >= 300) {
// Read error body immediately while stream is still open
String errorBody = readInputStream(response.body());
log.warn(
"HTTP request failed. URL: {} | Status: {} | Error:"
+ " {}",
request.getUrl(),
statusCode,
errorBody);
throw new CompletionException(
new HttpTransportException(
"HTTP request failed with status "
+ statusCode
+ " | "
+ errorBody,
statusCode,
errorBody));
var jdkRequest = buildJdkRequest(request, true);

return Flux.defer(
() -> {
AtomicReference<InputStream> responseBody = new AtomicReference<>();
return sendInputStreamAsync(jdkRequest, responseBody)
.timeout(Mono.delay(streamResponseTimeout()))
.flatMapMany(
response ->
handleStreamResponse(response, request, responseBody))
.doFinally(signal -> closeQuietly(responseBody.getAndSet(null)))
.onErrorMap(this::mapStreamError);
});
}

Comment thread
jujn marked this conversation as resolved.
private Mono<java.net.http.HttpResponse<InputStream>> sendInputStreamAsync(
java.net.http.HttpRequest request, AtomicReference<InputStream> responseBody) {
return Mono.create(
sink -> {
AtomicBoolean cancelled = new AtomicBoolean(false);
var future = client.sendAsync(request, BodyHandlers.ofInputStream());
sink.onCancel(
() -> {
cancelled.set(true);
future.cancel(true);
closeQuietly(responseBody.getAndSet(null));
});
future.whenComplete(
(response, error) -> {
if (error != null) {
if (!cancelled.get()) {
sink.error(error);
}
return response;
});

return Mono.fromCompletionStage(future)
.flatMapMany(response -> processStreamResponse(response, request))
.publishOn(Schedulers.boundedElastic())
.onErrorMap(
e -> !(e instanceof HttpTransportException),
e -> {
Throwable cause = e instanceof CompletionException ? e.getCause() : e;
if (cause instanceof HttpTransportException) {
return (HttpTransportException) cause;
}
return new HttpTransportException(
"SSE/NDJSON stream failed: " + e.getMessage(), e);
})
.subscribeOn(Schedulers.boundedElastic());
return;
}

responseBody.set(response.body());
if (cancelled.get()) {
closeQuietly(responseBody.getAndSet(null));
return;
}
sink.success(response);
});
});
}

private Flux<String> processStreamResponse(
java.net.http.HttpResponse<InputStream> response, HttpRequest request) {
private Flux<String> handleStreamResponse(
java.net.http.HttpResponse<InputStream> response,
HttpRequest request,
AtomicReference<InputStream> responseBody) {
InputStream inputStream = response.body();
responseBody.set(inputStream);

int statusCode = response.statusCode();
if (statusCode < 200 || statusCode >= 300) {
return readInputStreamAsync(inputStream)
.timeout(streamResponseTimeout())
.onErrorReturn("")
.flatMapMany(
errorBody -> {
log.warn(
Comment thread
jujn marked this conversation as resolved.
"HTTP request failed. URL: {} | Status: {} | Error: {}",
request.getUrl(),
statusCode,
errorBody);
return Flux.error(
new HttpTransportException(
"HTTP request failed with status "
+ statusCode
+ " | "
+ errorBody,
statusCode,
errorBody));
});
}

return processStreamResponse(inputStream, request)
.timeout(
// Timeout strategy 1: Time To First Token (TTFT).
// The maximum time to wait for the first piece of data after headers.
Mono.delay(streamResponseTimeout()),

// Timeout strategy 2: Inter-token gap (Stream Idle Timeout).
// The maximum time to wait between receiving two consecutive data chunks.
data -> Mono.delay(streamIdleTimeout()));
}

private Flux<String> processStreamResponse(InputStream inputStream, HttpRequest request) {
if (inputStream == null) {
return Flux.empty();
}
Expand All @@ -253,11 +294,13 @@ private Flux<String> processStreamResponse(

// Use Flux.using to manage resource lifecycle
return Flux.using(
() ->
new BufferedReader(
new InputStreamReader(inputStream, StandardCharsets.UTF_8)),
reader -> isNdjson ? readNdJsonLines(reader) : readSseLines(reader),
this::closeQuietly);
() ->
new BufferedReader(
new InputStreamReader(inputStream, StandardCharsets.UTF_8)),
reader -> isNdjson ? readNdJsonLines(reader) : readSseLines(reader),
this::closeQuietly)
// reader.lines() uses blocking I/O internally
.subscribeOn(Schedulers.boundedElastic());
}

private Flux<String> readSseLines(BufferedReader reader) {
Expand Down Expand Up @@ -310,16 +353,19 @@ public boolean isClosed() {
return closed.get();
}

private java.net.http.HttpRequest buildJdkRequest(HttpRequest request) {
private java.net.http.HttpRequest buildJdkRequest(HttpRequest request, boolean isStreaming) {
URI uri;
try {
uri = URI.create(request.getUrl());
} catch (IllegalArgumentException e) {
throw new HttpTransportException("Invalid URL: " + request.getUrl(), e);
}

var builder =
java.net.http.HttpRequest.newBuilder().uri(uri).timeout(config.getReadTimeout());
var builder = java.net.http.HttpRequest.newBuilder().uri(uri);

if (!isStreaming && config.getReadTimeout() != null) {
builder.timeout(config.getReadTimeout());
}

for (Map.Entry<String, String> header : request.getHeaders().entrySet()) {
builder.header(header.getKey(), header.getValue());
Expand Down Expand Up @@ -382,6 +428,38 @@ private String readInputStream(InputStream inputStream) {
}
}

private Mono<String> readInputStreamAsync(InputStream inputStream) {
return Mono.fromCallable(() -> readInputStream(inputStream))
.defaultIfEmpty("")
.subscribeOn(Schedulers.boundedElastic());
}

private Duration streamResponseTimeout() {
return config.getResponseTimeout() != null
? config.getResponseTimeout()
: HttpTransportConfig.DEFAULT_RESPONSE_TIMEOUT;
}

private Duration streamIdleTimeout() {
return config.getStreamIdleTimeout() != null
? config.getStreamIdleTimeout()
: HttpTransportConfig.DEFAULT_STREAM_IDLE_TIMEOUT;
}

private Throwable mapStreamError(Throwable e) {
if (e instanceof TimeoutException) {
return new HttpTransportException("Stream timeout: " + e.getMessage(), e);
}
if (e instanceof HttpTransportException) {
return e;
}
Throwable cause = e instanceof CompletionException ? e.getCause() : e;
if (cause instanceof HttpTransportException) {
return cause;
}
return new HttpTransportException("SSE/NDJSON stream failed: " + e.getMessage(), e);
}

private void closeQuietly(AutoCloseable closeable) {
if (closeable != null) {
try {
Expand Down
Loading
Loading