wanjia9506
/
langchain4j
mirror of https://github.com/langchain4j/langchain4j.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: do not consume the response body on the OkHttp streaming path (#5806) 

## Issue
Closes #5804

## Change

The streaming path used `fromOkHttpResponse(response)`, the converter
written for the synchronous path. That converter calls
`response.body().bytes()` unless the `Content-Type` is
`text/event-stream`, which reads the body to the end and closes it. The
parser then reads the same body, fails with `IOException: closed`, and
the listener receives no events. Servers that stream with another media
type hit this, for example Ollama with `application/x-ndjson`.

`fromOkHttpResponse` is now split into an overload that takes the body
bytes, and the streaming path passes `null`, the same way
`JdkHttpClient` calls `fromJdkResponse(jdkResponse, null)`.

```java
SuccessfulHttpResponse successResponse = fromOkHttpResponse(response, null);
```

The synchronous path is unchanged, and non-2xx responses are still
handled earlier by `readBody(response)`.

`OkHttpClientStreamingTest` serves responses from
`com.sun.net.httpserver.HttpServer`, so it needs no API key. It covers
the streaming path with `application/x-ndjson`, with
`text/event-stream`, with no `Content-Type`, and with a 500 response,
plus the synchronous path with and without `text/event-stream`.
Reverting the fix makes the two non-SSE streaming tests fail with
`IOException: closed`.

The import reordering and line-wrapping in the diff come from
`spotless:apply` (palantir), which reformats the whole file once it is
touched.

## General checklist
- [X] There are no breaking changes (API, behaviour)
- [X] I have added unit and/or integration tests for my change
- [X] The tests cover both positive and negative cases
- [X] I have manually run all the unit and integration tests in the
module I have added/changed, and they are all green
<!-- The two *IT classes in this module (OkHttpClientIT,
OkHttpClientTimeoutIT) require OPENAI_API_KEY and were not run. -->
- [ ] I have manually run all the unit and integration tests in the
[core](https://github.com/langchain4j/langchain4j/tree/main/langchain4j-core)
and
[main](https://github.com/langchain4j/langchain4j/tree/main/langchain4j)
modules, and they are all green
<!-- langchain4j-core was built and tested as part of the -am build and
is green; the langchain4j main module was not run. -->
- [ ] I have added/updated the
[documentation](https://github.com/langchain4j/langchain4j/tree/main/docs/docs)
<!-- Behaviour fix, no documentation change. -->
- [ ] I have added an example in the [examples
repo](https://github.com/langchain4j/langchain4j-examples) (only for
"big" features)
<!-- Bug fix. -->
- [ ] I have added/updated [Spring Boot
starter(s)](https://github.com/langchain4j/langchain4j-spring) (if
applicable)
<!-- No starter is affected. -->
This commit is contained in:
Eunbin Son 2026-07-20 16:53:11 +09:00 committed by GitHub
parent c27e6e9c2d
commit 1023c67539
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 223 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
http-clients/langchain4j-http-client-okhttp/src/main/java/dev/langchain4j/http/client/okhttp/OkHttpClient.java
View File

@ -1,5 +1,8 @@
package dev.langchain4j.http.client.okhttp;
import static dev.langchain4j.http.client.sse.ServerSentEventListenerUtils.ignoringExceptions;
import static dev.langchain4j.internal.Utils.getOrDefault;
import dev.langchain4j.exception.HttpException;
import dev.langchain4j.exception.TimeoutException;
import dev.langchain4j.http.client.FormDataFile;
@ -8,14 +11,6 @@ import dev.langchain4j.http.client.HttpRequest;
import dev.langchain4j.http.client.SuccessfulHttpResponse;
import dev.langchain4j.http.client.sse.ServerSentEventListener;
import dev.langchain4j.http.client.sse.ServerSentEventParser;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.MediaType;
import okhttp3.MultipartBody;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import java.io.IOException;
import java.io.InputStream;
import java.net.SocketTimeoutException;
@ -23,9 +18,13 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static dev.langchain4j.http.client.sse.ServerSentEventListenerUtils.ignoringExceptions;
import static dev.langchain4j.internal.Utils.getOrDefault;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.MediaType;
import okhttp3.MultipartBody;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
public class OkHttpClient implements HttpClient {
@ -77,7 +76,7 @@ public class OkHttpClient implements HttpClient {
return;
}
SuccessfulHttpResponse successResponse = fromOkHttpResponse(response);
SuccessfulHttpResponse successResponse = fromOkHttpResponse(response, null);
ignoringExceptions(() -> listener.onOpen(successResponse));
try (InputStream inputStream = getInputStream(response)) {
@ -107,11 +106,6 @@ public class OkHttpClient implements HttpClient {
}
private SuccessfulHttpResponse fromOkHttpResponse(Response response) throws IOException {
Map<String, List<String>> headers = new HashMap<>();
for (String name : response.headers().names()) {
headers.put(name, response.headers().values(name));
}
String contentType = response.header("content-type");
byte[] body;
if (contentType != null && contentType.contains("text/event-stream")) {
@ -120,6 +114,19 @@ public class OkHttpClient implements HttpClient {
body = response.body().bytes();
}
return fromOkHttpResponse(response, body);
}
/**
* Converts an OkHttp response into a {@link SuccessfulHttpResponse} without touching the response body.
* The streaming path passes {@code null} here, so that the body is left for the SSE parser to read.
*/
private SuccessfulHttpResponse fromOkHttpResponse(Response response, byte[] body) {
Map<String, List<String>> headers = new HashMap<>();
for (String name : response.headers().names()) {
headers.put(name, response.headers().values(name));
}
return SuccessfulHttpResponse.builder()
.statusCode(response.code())
.headers(headers)
@ -165,8 +172,7 @@ public class OkHttpClient implements HttpClient {
private RequestBody buildRequestBody(HttpRequest request) {
if (!request.formDataFields().isEmpty() || !request.formDataFiles().isEmpty()) {
MultipartBody.Builder multipartBuilder =
new MultipartBody.Builder().setType(MultipartBody.FORM);
MultipartBody.Builder multipartBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM);
for (Map.Entry<String, String> entry : request.formDataFields().entrySet()) {
multipartBuilder.addFormDataPart(entry.getKey(), entry.getValue());
@ -174,8 +180,7 @@ public class OkHttpClient implements HttpClient {
for (Map.Entry<String, FormDataFile> entry : request.formDataFiles().entrySet()) {
FormDataFile file = entry.getValue();
RequestBody fileBody = RequestBody.create(
file.content(), MediaType.parse(file.contentType()));
RequestBody fileBody = RequestBody.create(file.content(), MediaType.parse(file.contentType()));
multipartBuilder.addFormDataPart(entry.getKey(), file.fileName(), fileBody);
}

197
http-clients/langchain4j-http-client-okhttp/src/test/java/dev/langchain4j/http/client/okhttp/OkHttpClientStreamingTest.java Normal file
View File

@ -0,0 +1,197 @@
package dev.langchain4j.http.client.okhttp;
import static org.assertj.core.api.Assertions.assertThat;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import dev.langchain4j.exception.HttpException;
import dev.langchain4j.http.client.HttpMethod;
import dev.langchain4j.http.client.HttpRequest;
import dev.langchain4j.http.client.SuccessfulHttpResponse;
import dev.langchain4j.http.client.sse.ServerSentEvent;
import dev.langchain4j.http.client.sse.ServerSentEventListener;
import dev.langchain4j.http.client.sse.ServerSentEventParser;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
/**
* Verifies that the streaming path does not consume the response body before the parser reads it,
* regardless of the {@code Content-Type} the server sends.
*/
class OkHttpClientStreamingTest {
private static final List<String> LINES = List.of("{\"n\":0}", "{\"n\":1}", "{\"n\":2}");
private HttpServer server;
private String baseUrl;
@BeforeEach
void startServer() throws IOException {
server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
server.start();
baseUrl = "http://localhost:" + server.getAddress().getPort();
}
@AfterEach
void stopServer() {
server.stop(0);
}
@Test
void should_deliver_all_events_when_content_type_is_not_event_stream() throws Exception {
respondWith("application/x-ndjson", String.join("\n", LINES));
RecordingListener listener = streamFrom("/stream");
assertThat(listener.errors).isEmpty();
assertThat(listener.events).containsExactlyElementsOf(LINES);
assertThat(listener.response.get().body()).isNull();
}
@Test
void should_deliver_all_events_when_content_type_is_event_stream() throws Exception {
respondWith("text/event-stream", String.join("\n", LINES));
RecordingListener listener = streamFrom("/stream");
assertThat(listener.errors).isEmpty();
assertThat(listener.events).containsExactlyElementsOf(LINES);
assertThat(listener.response.get().body()).isNull();
}
@Test
void should_deliver_all_events_when_content_type_is_missing() throws Exception {
respondWith(null, String.join("\n", LINES));
RecordingListener listener = streamFrom("/stream");
assertThat(listener.errors).isEmpty();
assertThat(listener.events).containsExactlyElementsOf(LINES);
assertThat(listener.response.get().body()).isNull();
}
@Test
void should_notify_listener_about_error_when_response_is_not_successful() throws Exception {
server.createContext("/stream", exchange -> {
byte[] bytes = "server is down".getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().add("Content-Type", "application/x-ndjson");
exchange.sendResponseHeaders(500, bytes.length);
try (OutputStream outputStream = exchange.getResponseBody()) {
outputStream.write(bytes);
}
});
RecordingListener listener = streamFrom("/stream");
assertThat(listener.events).isEmpty();
assertThat(listener.response.get()).isNull();
assertThat(listener.errors).singleElement().isInstanceOfSatisfying(HttpException.class, exception -> {
assertThat(exception.statusCode()).isEqualTo(500);
assertThat(exception.getMessage()).isEqualTo("server is down");
});
}
@Test
void should_read_body_in_non_streaming_path() throws Exception {
respondWith("application/x-ndjson", String.join("\n", LINES));
SuccessfulHttpResponse response = OkHttpClient.builder().build().execute(request("/stream"));
assertThat(response.body()).isEqualTo(String.join("\n", LINES));
}
@Test
void should_not_read_body_in_non_streaming_path_when_content_type_is_event_stream() throws Exception {
respondWith("text/event-stream", String.join("\n", LINES));
SuccessfulHttpResponse response = OkHttpClient.builder().build().execute(request("/stream"));
assertThat(response.body()).isNull();
}
private void respondWith(String contentType, String body) {
server.createContext("/stream", (HttpExchange exchange) -> {
byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
if (contentType != null) {
exchange.getResponseHeaders().add("Content-Type", contentType);
}
exchange.sendResponseHeaders(200, bytes.length);
try (OutputStream outputStream = exchange.getResponseBody()) {
outputStream.write(bytes);
}
});
}
private RecordingListener streamFrom(String path) throws InterruptedException {
RecordingListener listener = new RecordingListener();
OkHttpClient.builder().build().execute(request(path), new LineParser(), listener);
assertThat(listener.completed.await(30, TimeUnit.SECONDS)).isTrue();
return listener;
}
private HttpRequest request(String path) {
return HttpRequest.builder().method(HttpMethod.GET).url(baseUrl + path).build();
}
/**
* Emits one event per line, like the parsers used for newline-delimited JSON streams.
*/
private static class LineParser implements ServerSentEventParser {
@Override
public void parse(InputStream httpResponseBody, ServerSentEventListener listener) {
try (BufferedReader reader =
new BufferedReader(new InputStreamReader(httpResponseBody, StandardCharsets.UTF_8))) {
String line;
while ((line = reader.readLine()) != null) {
listener.onEvent(new ServerSentEvent(null, line));
}
} catch (IOException e) {
listener.onError(e);
}
}
}
private static class RecordingListener implements ServerSentEventListener {
final AtomicReference<SuccessfulHttpResponse> response = new AtomicReference<>();
final List<String> events = new ArrayList<>();
final List<Throwable> errors = new CopyOnWriteArrayList<>();
final CountDownLatch completed = new CountDownLatch(1);
@Override
public void onOpen(SuccessfulHttpResponse response) {
this.response.set(response);
}
@Override
public void onEvent(ServerSentEvent event) {
events.add(event.data());
}
@Override
public void onError(Throwable throwable) {
errors.add(throwable);
completed.countDown();
}
@Override
public void onClose() {
completed.countDown();
}
}
}