feat(core): expose request body content streams

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 <reified T> 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<InputStream>()
+
+        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`.

openai-java-core/src/main/kotlin/com/openai/core/http/HttpRequestBody.kt

@@ -1,12 +1,27 @@
 package com.openai.core.http
 
+import java.io.ByteArrayOutputStream
+import java.io.InputStream
 import java.io.OutputStream
 import java.lang.AutoCloseable
 
 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

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.

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);
+        }
+    }
+}

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<String>()
+                            .value("value")
+                            .contentType("text/plain")
+                            .build(),
+                    "binary" to
+                        MultipartField.builder<ByteArray>()
+                            .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<java.io.InputStream>()
+                            .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")
+    }
+}

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<HttpResponse> =
+                    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()