misc: catch up to main (#1205)

* misc: enhance support for replayable instances of `InputStream` (#1197)

* chore: release 1.3.31

* chore: bump snapshot version to 1.3.32-SNAPSHOT

* fix: CBOR protocol test assertions / blob serialization (#1198)

* fix: correctly serialize subset of shape's members when configured (#1199)

* chore: release 1.3.32

* chore: bump snapshot version to 1.3.33-SNAPSHOT

---------

Co-authored-by: Matas <[email protected]>
Co-authored-by: aws-sdk-kotlin-ci <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## [1.3.32] - 01/06/2025
+
+### Fixes
+* Fix serialization of CBOR blobs
+
+## [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

codegen/smithy-aws-kotlin-codegen/src/main/kotlin/software/amazon/smithy/kotlin/codegen/aws/protocols/core/QueryHttpBindingProtocolGenerator.kt

@@ -145,7 +145,7 @@ abstract class AbstractQueryFormUrlSerializerGenerator(
         return shape.documentSerializer(ctx.settings, symbol, members) { writer ->
             writer.openBlock("internal fun #identifier.name:L(serializer: #T, input: #T) {", RuntimeTypes.Serde.Serializer, symbol)
                 .call {
-                    renderSerializerBody(ctx, shape, shape.members().toList(), writer)
+                    renderSerializerBody(ctx, shape, members.toList(), writer)
                 }
                 .closeBlock("}")
         }

codegen/smithy-aws-kotlin-codegen/src/test/kotlin/software/amazon/smithy/kotlin/codegen/aws/protocols/RpcV2CborTest.kt

@@ -4,6 +4,7 @@
  */
 package software.amazon.smithy.kotlin.codegen.aws.protocols
 
+import io.kotest.matchers.string.shouldNotContain
 import software.amazon.smithy.kotlin.codegen.test.*
 import kotlin.test.Test
 
@@ -145,4 +146,20 @@ class RpcV2CborTest {
         val serializeBody = serializer.lines("    override suspend fun serialize(context: ExecutionContext, input: PutFooStreamingRequest): HttpRequestBuilder {", "}")
         serializeBody.shouldContainOnlyOnceWithDiff("""builder.headers.setMissing("Content-Type", "application/vnd.amazon.eventstream")""")
     }
+
+    @Test
+    fun testEventStreamInitialRequestDoesNotSerializeStreamMember() {
+        val ctx = model.newTestContext("CborExample")
+
+        val generator = RpcV2Cbor()
+        generator.generateProtocolClient(ctx.generationCtx)
+
+        ctx.generationCtx.delegator.finalize()
+        ctx.generationCtx.delegator.flushWriters()
+
+        val documentSerializer = ctx.manifest.expectFileString("/src/main/kotlin/com/test/serde/PutFooStreamingRequestDocumentSerializer.kt")
+
+        val serializeBody = documentSerializer.lines("    serializer.serializeStruct(OBJ_DESCRIPTOR) {", "}")
+        serializeBody.shouldNotContain("input.messages") // `messages` is the stream member and should not be serialized in the initial request
+    }
 }

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/CborSerializerGenerator.kt

@@ -108,7 +108,7 @@ class CborSerializerGenerator(
         val symbol = ctx.symbolProvider.toSymbol(shape)
         return shape.documentSerializer(ctx.settings, symbol, members) { writer ->
             writer.withBlock("internal fun #identifier.name:L(serializer: #T, input: #T) {", "}", RuntimeTypes.Serde.Serializer, symbol) {
-                call { renderSerializerBody(ctx, shape, shape.members().toList(), writer) }
+                call { renderSerializerBody(ctx, shape, members.toList(), writer) }
             }
         }
     }

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.31-SNAPSHOT
+sdkVersion=1.3.33-SNAPSHOT
 
 # codegen
-codegenVersion=0.33.31-SNAPSHOT
+codegenVersion=0.33.33-SNAPSHOT

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