Merge branch 'main' of https://github.com/awslabs/smithy-kotlin into …

…flexible-checksums

.changes/1021e75a-45f3-4f3a-820c-700d9ec6e782.json

@@ -0,0 +1,5 @@
+{
+    "id": "1021e75a-45f3-4f3a-820c-700d9ec6e782",
+    "type": "bugfix",
+    "description": "Fix serialization of CBOR blobs"
+}

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [1.3.31] - 12/18/2024
+
+### Features
+* [#1473](https://github.com/awslabs/aws-sdk-kotlin/issues/1473) Enhance support for replayable instances of `InputStream`
+
+## [1.3.30] - 12/16/2024
+
 ## [1.3.29] - 12/12/2024
 
 ## [1.3.28] - 12/03/2024

codegen/smithy-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/rendering/protocol/HttpProtocolUnitTestRequestGenerator.kt

@@ -117,11 +117,11 @@ open class HttpProtocolUnitTestRequestGenerator protected constructor(builder: B
                         write("return")
                     }
                     write("requireNotNull(expectedBytes) { #S }", "expected application/cbor body cannot be null")
-                    write("requireNotNull(expectedBytes) { #S }", "actual application/cbor body cannot be null")
+                    write("requireNotNull(actualBytes) { #S }", "actual application/cbor body cannot be null")
 
                     write("")
                     write("val expectedRequest = #L(#T(expectedBytes))", inputDeserializer.name, RuntimeTypes.Serde.SerdeCbor.CborDeserializer)
-                    write("val actualRequest = #L(#T(expectedBytes))", inputDeserializer.name, RuntimeTypes.Serde.SerdeCbor.CborDeserializer)
+                    write("val actualRequest = #L(#T(actualBytes))", inputDeserializer.name, RuntimeTypes.Serde.SerdeCbor.CborDeserializer)
                     write("assertEquals(expectedRequest, actualRequest)")
                 }
                 writer.write("")

codegen/smithy-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/rendering/serde/SerializeStructGenerator.kt

@@ -647,7 +647,6 @@ open class SerializeStructGenerator(
         val target = member.targetOrSelf(ctx.model)
 
         val encoded = when {
-            target.type == ShapeType.BLOB -> writer.format("#L.#T()", identifier, RuntimeTypes.Core.Text.Encoding.encodeBase64String)
             target.type == ShapeType.TIMESTAMP -> {
                 writer.addImport(RuntimeTypes.Core.TimestampFormat)
                 val tsFormat = member

codegen/smithy-kotlin-codegen/src/test/kotlin/software/amazon/smithy/kotlin/codegen/rendering/serde/SerializeStructGeneratorTest.kt

@@ -1822,7 +1822,7 @@ class SerializeStructGeneratorTest {
 
         val expected = """
             serializer.serializeStruct(OBJ_DESCRIPTOR) {
-                input.fooBlob?.let { field(FOOBLOB_DESCRIPTOR, it.encodeBase64String()) }
+                input.fooBlob?.let { field(FOOBLOB_DESCRIPTOR, it) }
             }
         """.trimIndent()
 

gradle.properties

@@ -13,7 +13,7 @@ kotlinx.atomicfu.enableNativeIrTransformation=false
 org.gradle.jvmargs=-Xmx2G -XX:MaxMetaspaceSize=1G
 
 # SDK
-sdkVersion=1.3.30-SNAPSHOT
+sdkVersion=1.3.32-SNAPSHOT
 
 # codegen
-codegenVersion=0.33.30-SNAPSHOT
+codegenVersion=0.33.32-SNAPSHOT

runtime/runtime-core/common/src/aws/smithy/kotlin/runtime/util/JMESPath.kt

@@ -66,10 +66,20 @@ public fun Any?.type(): String = when (this) {
     else -> throw Exception("Undetected type for: $this")
 }
 
+// Collection `flattenIfPossible` functions
 @InternalApi
 @JvmName("noOpUnnestedCollection")
 public inline fun <reified T> Collection<T>.flattenIfPossible(): Collection<T> = this
 
 @InternalApi
 @JvmName("flattenNestedCollection")
 public inline fun <reified T> Collection<Collection<T>>.flattenIfPossible(): Collection<T> = flatten()
+
+// List `flattenIfPossible` functions
+@InternalApi
+@JvmName("noOpUnnestedCollection")
+public inline fun <reified T> List<T>.flattenIfPossible(): List<T> = this
+
+@InternalApi
+@JvmName("flattenNestedCollection")
+public inline fun <reified T> List<List<T>>.flattenIfPossible(): List<T> = flatten()

runtime/runtime-core/common/test/aws/smithy/kotlin/runtime/util/JmesPathTest.kt

@@ -0,0 +1,49 @@
+package aws.smithy.kotlin.runtime.util
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class JmesPathTest {
+    @Test
+    fun flattenNestedLists() {
+        val nestedList = listOf(
+            listOf(1, 2, 3),
+            listOf(4, 5),
+            listOf(6),
+        )
+        val flattenedList = nestedList.flattenIfPossible()
+        assertEquals(listOf(1, 2, 3, 4, 5, 6), flattenedList)
+    }
+
+    @Test
+    fun flattenEmptyNestedLists() {
+        val nestedList = listOf(
+            listOf<Int>(),
+            listOf(),
+            listOf(),
+        )
+        val flattenedList = nestedList.flattenIfPossible()
+        assertTrue(flattenedList.isEmpty())
+    }
+
+    @Test
+    fun flattenNestedEmptyAndNonEmptyNestedLists() {
+        val nestedList = listOf(
+            listOf(1, 2),
+            listOf(),
+            listOf(3, 4, 5),
+        )
+        val flattenedList = nestedList.flattenIfPossible()
+        assertEquals(listOf(1, 2, 3, 4, 5), flattenedList)
+    }
+
+    @Test
+    fun flattenList() {
+        val nestedList = listOf(
+            listOf(1, 2, 3),
+        )
+        val flattenedList = nestedList.flattenIfPossible()
+        assertEquals(listOf(1, 2, 3), flattenedList)
+    }
+}

runtime/runtime-core/jvm/src/aws/smithy/kotlin/runtime/content/ByteStreamJVM.kt

@@ -114,11 +114,30 @@ public fun ByteStream.Companion.fromInputStream(
  * @param contentLength If specified, indicates how many bytes remain in this stream. Defaults to `null`.
  */
 public fun InputStream.asByteStream(contentLength: Long? = null): ByteStream.SourceStream {
-    val source = source()
+    if (markSupported() && contentLength != null) {
+        mark(contentLength.toInt())
+    }
+
     return object : ByteStream.SourceStream() {
         override val contentLength: Long? = contentLength
         override val isOneShot: Boolean = !markSupported()
-        override fun readFrom(): SdkSource = source
+        override fun readFrom(): SdkSource {
+            if (markSupported() && contentLength != null) {
+                reset()
+                mark(contentLength.toInt())
+                return object : SdkSource by source() {
+                    /*
+                     * This is a no-op close to prevent body hashing from closing the underlying InputStream, which causes
+                     * `IOException: Stream closed` on subsequent reads. Consider making [ByteStream.ChannelStream]/[ByteStream.SourceStream]
+                     * (or possibly even [ByteStream] itself) implement [Closeable] to better handle closing streams.
+                     * This should allow us to clean up our usage of [ByteStream.cancel()].
+                     */
+                    override fun close() { }
+                }
+            }
+
+            return source()
+        }
     }
 }
 

runtime/runtime-core/jvm/test/aws/smithy/kotlin/runtime/content/ByteStreamJVMTest.kt

@@ -5,8 +5,11 @@
 
 package aws.smithy.kotlin.runtime.content
 
+import aws.smithy.kotlin.runtime.io.readToByteArray
 import aws.smithy.kotlin.runtime.testing.RandomTempFile
 import kotlinx.coroutines.test.runTest
+import java.io.BufferedInputStream
+import java.io.ByteArrayInputStream
 import java.io.ByteArrayOutputStream
 import java.io.InputStream
 import java.io.OutputStream
@@ -228,6 +231,31 @@ class ByteStreamJVMTest {
         assertFalse(sos.closed)
     }
 
+    // https://github.com/awslabs/aws-sdk-kotlin/issues/1473
+    @Test
+    fun testReplayableInputStreamAsByteStream() = runTest {
+        val content = "Hello, Bytes!".encodeToByteArray()
+        val byteArrayIns = ByteArrayInputStream(content)
+        val nonReplayableIns = NonReplayableInputStream(byteArrayIns)
+
+        // buffer the non-replayable stream, making it replayable...
+        val bufferedIns = BufferedInputStream(nonReplayableIns)
+
+        val byteStream = bufferedIns.asByteStream(content.size.toLong())
+
+        // Test that it can be read at least twice (e.g. once for hashing the body, once for transmitting the body)
+        assertContentEquals(content, byteStream.readFrom().use { it.readToByteArray() })
+        assertContentEquals(content, byteStream.readFrom().use { it.readToByteArray() })
+    }
+
+    private class NonReplayableInputStream(val inputStream: InputStream) : InputStream() {
+        override fun markSupported(): Boolean = false // not replayable
+
+        override fun read(): Int = inputStream.read()
+        override fun mark(readlimit: Int) = inputStream.mark(readlimit)
+        override fun reset() = inputStream.reset()
+    }
+
     private class StatusTrackingOutputStream(val os: OutputStream) : OutputStream() {
         var closed: Boolean = false
 

runtime/serde/serde-xml/common/src/aws/smithy/kotlin/runtime/serde/xml/XmlSerializer.kt

@@ -134,7 +134,7 @@ public class XmlSerializer(private val xmlWriter: XmlStreamWriter = xmlStreamWri
         field(descriptor, value.format(format))
 
     override fun field(descriptor: SdkFieldDescriptor, value: ByteArray): Unit =
-        field(descriptor, value)
+        field(descriptor, value.encodeBase64String())
 
     override fun field(descriptor: SdkFieldDescriptor, value: Document?): Unit = throw SerializationException(
         "cannot serialize field ${descriptor.serialName}; Document type is not supported by xml encoding",