From 1d41280f508ce85cf882086be3d2f6393d3bb654 Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Wed, 30 Sep 2026 23:15:39 +0100 Subject: [PATCH 1/2] feat(core): expose request body content streams Resolved against current multipart lifecycle handling. --- .../com/openai/core/http/HttpRequestBodies.kt | 34 ++++++++ .../com/openai/core/http/HttpRequestBody.kt | 15 ++++ .../core/http/HttpRequestBodyJavaTest.java | 54 +++++++++++++ .../core/http/HttpRequestBodyContentTest.kt | 78 +++++++++++++++++++ 4 files changed, 181 insertions(+) create mode 100644 openai-java-core/src/test/java/com/openai/core/http/HttpRequestBodyJavaTest.java create mode 100644 openai-java-core/src/test/kotlin/com/openai/core/http/HttpRequestBodyContentTest.kt diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBodies.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBodies.kt index 4f47010fc..7674f0d15 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBodies.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBodies.kt @@ -13,6 +13,7 @@ import com.openai.errors.OpenAIInvalidDataException import java.io.ByteArrayInputStream import java.io.InputStream import java.io.OutputStream +import java.io.SequenceInputStream import java.util.Collections import java.util.IdentityHashMap import java.util.UUID @@ -28,6 +29,8 @@ internal inline fun json(jsonMapper: JsonMapper, value: T): HttpRequ override fun writeTo(outputStream: OutputStream) = outputStream.write(bytes) + override fun content(): InputStream = bytes.inputStream() + override fun contentType(): String = "application/json" override fun contentLength(): Long = bytes.size.toLong() @@ -96,6 +99,8 @@ private fun buildMultipartBody( outputStream.write(byteArray) } + override fun content(): InputStream = byteArray.inputStream() + override fun contentType(): String = contentType override fun contentLength(): Long = byteArray.size.toLong() @@ -111,6 +116,8 @@ private fun buildMultipartBody( bytes.copyTo(outputStream) } + override fun content(): InputStream = bytes + override fun contentType(): String = contentType override fun contentLength(): Long = -1L @@ -225,6 +232,33 @@ private constructor( outputStream.write(CRLF) } + // This must remain in sync with `writeTo`. + override fun content(): InputStream { + val streams = mutableListOf() + + parts.forEach { part -> + streams.add(DASHDASH.inputStream()) + streams.add(boundaryBytes.inputStream()) + streams.add(CRLF.inputStream()) + streams.add(CONTENT_DISPOSITION.inputStream()) + streams.add(part.contentDisposition.toByteArray().inputStream()) + streams.add(CRLF.inputStream()) + streams.add(CONTENT_TYPE.inputStream()) + streams.add(part.contentType.toByteArray().inputStream()) + streams.add(CRLF.inputStream()) + streams.add(CRLF.inputStream()) + streams.add(part.body.content()) + streams.add(CRLF.inputStream()) + } + + streams.add(DASHDASH.inputStream()) + streams.add(boundaryBytes.inputStream()) + streams.add(DASHDASH.inputStream()) + streams.add(CRLF.inputStream()) + + return SequenceInputStream(Collections.enumeration(streams)) + } + override fun contentType(): String = contentType // This must remain in sync with `writeTo`. diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBody.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBody.kt index f44a762ca..1fd32b4df 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBody.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBody.kt @@ -1,5 +1,7 @@ package com.openai.core.http +import java.io.ByteArrayOutputStream +import java.io.InputStream import java.io.OutputStream import java.lang.AutoCloseable @@ -7,6 +9,19 @@ interface HttpRequestBody : AutoCloseable { fun writeTo(outputStream: OutputStream) + /** + * Returns the request body content as an input stream. + * + * The default implementation buffers the bytes produced by [writeTo] so existing third-party + * implementations remain compatible. Implementations backed by an existing byte array or stream + * should override this method to avoid buffering. + */ + fun content(): InputStream { + val outputStream = ByteArrayOutputStream() + writeTo(outputStream) + return outputStream.toByteArray().inputStream() + } + fun contentType(): String? fun contentLength(): Long diff --git a/openai-java-core/src/test/java/com/openai/core/http/HttpRequestBodyJavaTest.java b/openai-java-core/src/test/java/com/openai/core/http/HttpRequestBodyJavaTest.java new file mode 100644 index 000000000..5c35183b0 --- /dev/null +++ b/openai-java-core/src/test/java/com/openai/core/http/HttpRequestBodyJavaTest.java @@ -0,0 +1,54 @@ +package com.openai.core.http; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; + +final class HttpRequestBodyJavaTest { + + @Test + void contentIsJavaDefaultMethod() throws Exception { + assertThat(HttpRequestBody.class.getMethod("content").isDefault()).isTrue(); + + HttpRequestBody body = + new HttpRequestBody() { + @Override + public void writeTo(OutputStream outputStream) { + try { + outputStream.write("body".getBytes(StandardCharsets.UTF_8)); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + @Override + public String contentType() { + return "text/plain"; + } + + @Override + public long contentLength() { + return 4L; + } + + @Override + public boolean repeatable() { + return true; + } + + @Override + public void close() {} + }; + + try (InputStream content = body.content()) { + byte[] bytes = new byte[4]; + assertThat(content.read(bytes)).isEqualTo(bytes.length); + assertThat(bytes).isEqualTo("body".getBytes(StandardCharsets.UTF_8)); + assertThat(content.read()).isEqualTo(-1); + } + } +} diff --git a/openai-java-core/src/test/kotlin/com/openai/core/http/HttpRequestBodyContentTest.kt b/openai-java-core/src/test/kotlin/com/openai/core/http/HttpRequestBodyContentTest.kt new file mode 100644 index 000000000..38ee82c68 --- /dev/null +++ b/openai-java-core/src/test/kotlin/com/openai/core/http/HttpRequestBodyContentTest.kt @@ -0,0 +1,78 @@ +package com.openai.core.http + +import com.openai.core.MultipartField +import com.openai.core.jsonMapper +import java.io.ByteArrayOutputStream +import java.io.OutputStream +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test + +internal class HttpRequestBodyContentTest { + + @Test + fun content_defaultsToWriteTo() { + val body = + object : HttpRequestBody { + override fun writeTo(outputStream: OutputStream) { + outputStream.write("body".toByteArray()) + } + + override fun contentType(): String = "text/plain" + + override fun contentLength(): Long = 4L + + override fun repeatable(): Boolean = true + + override fun close() {} + } + + body.content().use { content -> assertThat(content.readBytes()).isEqualTo("body".toByteArray()) } + } + + @Test + fun multipartContent_matchesWriteTo() { + val body = + multipartFormData( + jsonMapper(), + mapOf( + "field" to + MultipartField.builder() + .value("value") + .contentType("text/plain") + .build(), + "binary" to + MultipartField.builder() + .value("abc".toByteArray()) + .contentType("application/octet-stream") + .build(), + ), + ) + + val output = ByteArrayOutputStream() + body.writeTo(output) + + body.content().use { content -> assertThat(content.readBytes()).isEqualTo(output.toByteArray()) } + } + + @Test + fun multipartContent_streamsInputStreamParts() { + val body = + multipartFormData( + jsonMapper(), + mapOf( + "data" to + MultipartField.builder() + .value("stream content".byteInputStream().buffered()) + .contentType("application/octet-stream") + .build() + ), + ) + + val content = body.content().use { it.readBytes().toString(Charsets.UTF_8) } + + assertThat(body.repeatable()).isFalse() + assertThat(content).contains("Content-Disposition: form-data; name=\"data\"") + assertThat(content).contains("Content-Type: application/octet-stream") + assertThat(content).contains("stream content") + } +} From 772192423f59af1d69b569fb3be5ad181412d645 Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Wed, 30 Sep 2026 23:37:02 +0100 Subject: [PATCH 2/2] fix(core): preserve streaming through request logging --- .../com/openai/core/http/LoggingHttpClient.kt | 51 ++++++++++++++++ .../openai/core/http/LoggingHttpClientTest.kt | 58 +++++++++++++++++++ 2 files changed, 109 insertions(+) diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/LoggingHttpClient.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/LoggingHttpClient.kt index 22b1fbe1a..901d4cc63 100644 --- a/openai-java-core/src/main/kotlin/com/openai/core/http/LoggingHttpClient.kt +++ b/openai-java-core/src/main/kotlin/com/openai/core/http/LoggingHttpClient.kt @@ -284,6 +284,9 @@ private class LoggingHttpRequestBody( System.err.println() } + override fun content(): InputStream = + LoggingRequestInputStream(method, body.content(), charset) + override fun contentType(): String? = body.contentType() override fun contentLength(): Long = body.contentLength() @@ -293,6 +296,54 @@ private class LoggingHttpRequestBody( override fun close() = body.close() } +private class LoggingRequestInputStream( + private val method: HttpMethod, + private val inputStream: InputStream, + charset: Charset?, +) : InputStream() { + + private var isDone = false + private val buffer = LoggingBuffer(charset) + + override fun read(): Int { + if (isDone) return -1 + val byte = inputStream.read() + if (byte == -1) { + markDone() + return -1 + } + buffer.write(byte) + return byte + } + + override fun read(bytes: ByteArray, offset: Int, length: Int): Int { + if (isDone) return -1 + val count = inputStream.read(bytes, offset, length) + if (count == -1) { + markDone() + return -1 + } + for (index in offset until offset + count) { + buffer.write(bytes[index].toInt() and 0xFF) + } + return count + } + + override fun close() { + if (!isDone) markDone(closedEarly = true) + inputStream.close() + } + + private fun markDone(closedEarly: Boolean = false) { + if (isDone) return + isDone = true + buffer.flush() + val suffix = if (closedEarly) ", closed early" else "" + System.err.println("--> END " + method + " (" + buffer.writeCount() + "-byte body" + suffix + ")") + System.err.println() + } +} + /** * An [OutputStream] wrapper that delegates to [outputStream] while also logging bytes line by line * as it's written to. diff --git a/openai-java-core/src/test/kotlin/com/openai/core/http/LoggingHttpClientTest.kt b/openai-java-core/src/test/kotlin/com/openai/core/http/LoggingHttpClientTest.kt index 14f6dbf4d..567efb4a7 100644 --- a/openai-java-core/src/test/kotlin/com/openai/core/http/LoggingHttpClientTest.kt +++ b/openai-java-core/src/test/kotlin/com/openai/core/http/LoggingHttpClientTest.kt @@ -848,6 +848,64 @@ internal class LoggingHttpClientTest { ) } + @Test + fun debugLevel_contentStreamsWithoutFallingBackToWriteTo() { + val bytes = "streaming body".toByteArray(StandardCharsets.UTF_8) + var contentCalls = 0 + val requestBody = + object : HttpRequestBody { + override fun writeTo(outputStream: OutputStream) { + error("writeTo must not be used by the content consumer") + } + + override fun content(): InputStream { + contentCalls++ + return bytes.inputStream() + } + + override fun contentType(): String = "text/plain" + + override fun contentLength(): Long = bytes.size.toLong() + + override fun repeatable(): Boolean = false + + override fun close() {} + } + val delegate = + object : HttpClient { + override fun execute( + request: HttpRequest, + requestOptions: RequestOptions, + ): HttpResponse { + request.body!!.content().use { + assertThat(it.readBytes()).isEqualTo(bytes) + } + return fakeResponse(200, Headers.builder().build(), ByteArray(0)) + } + + override fun executeAsync( + request: HttpRequest, + requestOptions: RequestOptions, + ): CompletableFuture = + CompletableFuture.completedFuture(execute(request, requestOptions)) + + override fun close() {} + } + val request = + HttpRequest.builder() + .method(HttpMethod.POST) + .baseUrl("https://api.example.com") + .addPathSegment("stream") + .body(requestBody) + .build() + val client = loggingClient(delegate, LogLevel.DEBUG) + + client.execute(request, false).close() + + assertThat(contentCalls).isEqualTo(1) + assertThat(stderrOutput()).contains("--> END POST (" + bytes.size + "-byte body)") + } + @Test fun builder_toBuilder_roundtrips() { val delegate = fakeHttpClient()