feat(core): add HTTP adapter headers and body accessors

.castiron.stats.yml

@@ -5,4 +5,4 @@ openapi_transformed_spec_hash: aec25c66453cfdadd072d47e262b4c42
 config_hash: 0adc48b9b2b37cb31efd58e02e66aaa8
 codegen_sha: 7d1ad3e0d53fd0742357551b1713d6c3f43fd61f
 codegen_hash: 9d4e0ad7cff3d5010866de62216e9d9d0e580556771cbf5b2fc5dc5f5956757b
-public_codegen_sha: 5136b8ece6dfb3b49a77e0fef1bfa105c2d90d87
+public_codegen_sha: c52d64182f1de943bb6d6725ad83a164d86a8cf1

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

@@ -21,8 +21,22 @@ private constructor(
 
     fun names(): Set<String> = map.keys
 
+    /**
+     * Returns all values associated with [name], or an empty list if the header is absent. Names
+     * returned by [names] always have at least one value.
+     */
     fun values(name: String): List<String> = map[name].orEmpty()
 
+    /**
+     * Returns an immutable map of header names to their values, with case-insensitive lookup. Value
+     * lists are immutable and preserve insertion order.
+     *
+     * ```java
+     * Map<String, List<String>> headers = request.headers().asMap();
+     * ```
+     */
+    fun asMap(): Map<String, List<String>> = map
+
     fun toBuilder(): Builder = Builder().putAll(map)
 
     companion object {

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

@@ -11,8 +11,10 @@ import com.openai.core.multipartValueToTree
 import com.openai.core.toImmutable
 import com.openai.errors.OpenAIInvalidDataException
 import java.io.ByteArrayInputStream
+import java.io.FilterInputStream
 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 +30,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 +100,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 +117,13 @@ private fun buildMultipartBody(
                                         bytes.copyTo(outputStream)
                                     }
 
+                                    override fun content(): InputStream =
+                                        object : FilterInputStream(bytes) {
+                                            // SequenceInputStream closes parts at EOF. A later
+                                            // part may share this upload; the aggregate owns it.
+                                            override fun close() {}
+                                        }
+
                                     override fun contentType(): String = contentType
 
                                     override fun contentLength(): Long = -1L
@@ -225,6 +238,39 @@ 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 object : SequenceInputStream(Collections.enumeration(streams)) {
+            override fun close() {
+                // Release uploads through the same close-once tracking used by the body and
+                // cancellation, including every remaining upload if a close throws.
+                AutoCloseable { closeAll(parts) { it.body.close() } }.use { super.close() }
+            }
+        }
+    }
+
     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,43 @@
 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 stream produces the same bytes as [writeTo]. Close the returned stream after each
+     * attempt, and close this body after the final attempt, including on failure or cancellation.
+     * Closing a stream may release its underlying upload resources. Always close the body too, even
+     * if opening or reading the stream fails.
+     *
+     * If [repeatable] is true, each call returns a fresh stream over equivalent content. Otherwise,
+     * consume the body only once, using either this method or [writeTo]. Concurrent consumption is
+     * not supported.
+     *
+     * 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.
+     *
+     * ```java
+     * try (InputStream content = body.content()) {
+     *     // Pass content to the HTTP transport and wait for consumption to finish.
+     * }
+     * // Close body after the final transport attempt.
+     * ```
+     */
+    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,8 @@ 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 +295,62 @@ 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 var isClosed = 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 (offset < 0 || length < 0 || offset > bytes.size - length) {
+            throw IndexOutOfBoundsException()
+        }
+        if (length == 0) return 0
+        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 (isClosed) return
+        isClosed = true
+        inputStream.use { if (!isDone) markDone(closedEarly = true) }
+    }
+
+    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/kotlin/com/openai/core/http/HeadersMapViewTest.kt

@@ -0,0 +1,45 @@
+package com.openai.core.http
+
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.Test
+
+internal class HeadersMapViewTest {
+    @Test
+    fun preservesNamesAndOrderedValuesWithCaseInsensitiveLookup() {
+        val headers = Headers.builder().put("X-Test", "one").put("x-test", "two").build()
+        val map = headers.asMap()
+
+        assertThat(map).hasSize(1)
+        assertThat(map.keys).containsExactly("X-Test")
+        assertThat(map["X-TEST"]).containsExactly("one", "two")
+        assertThat(map.containsKey("x-test")).isTrue()
+        assertThat(map["missing"]).isNull()
+        assertThat(headers.values("missing")).isEmpty()
+        assertThat(Headers.builder().build().asMap()).isEmpty()
+    }
+
+    @Test
+    fun rejectsMapAndValueMutations() {
+        val map = Headers.builder().put("X-Test", "one").build().asMap() as MutableMap
+
+        assertThatThrownBy { map["Other"] = listOf("value") }
+            .isInstanceOf(UnsupportedOperationException::class.java)
+        assertThatThrownBy { (map.getValue("x-test") as MutableList).add("two") }
+            .isInstanceOf(UnsupportedOperationException::class.java)
+        assertThatThrownBy { map.entries.first().setValue(listOf("two")) }
+            .isInstanceOf(UnsupportedOperationException::class.java)
+    }
+
+    @Test
+    fun remainsUnchangedWhenBuildersAreModified() {
+        val builder = Headers.builder().put("X-Test", "one")
+        val headers = builder.build()
+        val map = headers.asMap()
+
+        builder.put("x-test", "two")
+        headers.toBuilder().replace("X-Test", "three").build()
+
+        assertThat(map["x-test"]).containsExactly("one")
+    }
+}