feat!: implement client config overrides for one or more operations (#776)

Author: lucix-aws

.changes/6d4356bb-d28e-45e9-b025-7adcfa94db68.json

@@ -0,0 +1,5 @@
+{
+    "id": "6d4356bb-d28e-45e9-b025-7adcfa94db68",
+    "type": "misc",
+    "description": "**Breaking** Remove `Closeable` supertype from `HttpClientEngine` interface. See [this discussion post](https://github.com/awslabs/aws-sdk-kotlin/discussions/818) for more information."
+}

.changes/f071b519-bd67-402b-826a-ced0a5b4c20d.json

@@ -0,0 +1,5 @@
+{
+    "id": "f071b519-bd67-402b-826a-ced0a5b4c20d",
+    "type": "feature",
+    "description": "Allow config override for one or more operations with an existing service client."
+}

runtime/auth/aws-credentials/common/src/aws/smithy/kotlin/runtime/auth/awscredentials/CredentialsProvider.kt

@@ -4,6 +4,8 @@
  */
 package aws.smithy.kotlin.runtime.auth.awscredentials
 
+import aws.smithy.kotlin.runtime.io.Closeable
+
 /**
  * Represents a producer/source of AWS credentials
  */
@@ -13,3 +15,11 @@ public interface CredentialsProvider {
      */
     public suspend fun getCredentials(): Credentials
 }
+
+/**
+ * A [CredentialsProvider] with [Closeable] resources. Users SHOULD call [close] when done with the provider to ensure
+ * any held resources are properly released.
+ *
+ * Implementations SHOULD evict any previously-retrieved or stored credentials when the provider is closed.
+*/
+public interface CloseableCredentialsProvider : CredentialsProvider, Closeable

runtime/auth/aws-signing-tests/common/src/aws/smithy/kotlin/runtime/auth/awssigning/tests/MiddlewareSigningTestBase.kt

@@ -88,7 +88,7 @@ public abstract class MiddlewareSigningTestBase : HasSigner {
                 return HttpCall(request, resp, now, now)
             }
         }
-        val client = sdkHttpClient(mockEngine)
+        val client = SdkHttpClient(mockEngine)
 
         operation.roundTrip(client, Unit)
         return operation.context[HttpOperationContext.HttpCallList].last().request

runtime/auth/aws-signing-tests/jvm/src/aws/smithy/kotlin/runtime/auth/awssigning/tests/SigningSuiteTestBaseJVM.kt

@@ -259,7 +259,7 @@ public actual abstract class SigningSuiteTestBase : HasSigner {
                 return HttpCall(request, resp, now, now)
             }
         }
-        val client = sdkHttpClient(mockEngine)
+        val client = SdkHttpClient(mockEngine)
 
         operation.roundTrip(client, Unit)
         return operation.context[HttpOperationContext.HttpCallList].last().request

runtime/io/common/src/aws/smithy/kotlin/runtime/io/Closeable.kt

@@ -5,6 +5,8 @@
 
 package aws.smithy.kotlin.runtime.io
 
+import aws.smithy.kotlin.runtime.util.InternalApi
+
 // this really should live in the stdlib...
 // https://youtrack.jetbrains.com/issue/KT-31066
 
@@ -43,3 +45,8 @@ public inline fun <C : Closeable, R> C.use(block: (C) -> R): R {
         }
     }
 }
+
+@InternalApi
+public fun Any.closeIfCloseable() {
+    if (this is Closeable) close()
+}

runtime/io/common/src/aws/smithy/kotlin/runtime/io/SdkManaged.kt

@@ -0,0 +1,56 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.smithy.kotlin.runtime.io
+
+import aws.smithy.kotlin.runtime.util.InternalApi
+import kotlinx.atomicfu.locks.SynchronizedObject
+import kotlinx.atomicfu.locks.synchronized
+
+/**
+ * Interface that the lifecycle of some resource is managed by the SDK at runtime.
+ */
+@InternalApi
+public interface SdkManaged {
+    /**
+     * Invoked by a caller to declare usership of the resource.
+     */
+    public fun share()
+
+    /**
+     * Invoked when a caller releases the resource. Returns a boolean indicating whether the resource has been fully
+     * unshared with this call (i.e. the caller was the only remaining user). Future calls on a fully unshared object
+     * would return false.
+     */
+    public fun unshare(): Boolean
+}
+
+/**
+ * Abstract class which implements usage count tracking for [SdkManaged].
+ */
+@InternalApi
+public abstract class SdkManagedBase : SdkManaged {
+    private val state = object : SynchronizedObject() {
+        var shareCount: Int = 0
+        var isUnshared: Boolean = false
+    }
+
+    override fun share() {
+        synchronized(state) {
+            check(!state.isUnshared) { "caller attempted to share() a fully unshared object" }
+
+            state.shareCount++
+        }
+    }
+
+    override fun unshare(): Boolean = synchronized(state) {
+        if (state.isUnshared) return false
+
+        state.shareCount--
+        if (state.shareCount > 0) return false
+
+        state.isUnshared = true
+        true
+    }
+}

runtime/io/common/src/aws/smithy/kotlin/runtime/io/SdkManagedCloseable.kt

@@ -0,0 +1,20 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.smithy.kotlin.runtime.io
+
+import aws.smithy.kotlin.runtime.util.InternalApi
+
+/**
+ * Wraps a [Closeable] to operate as an [SdkManaged] with share count tracking.
+ * The final [unshare] call will trigger the closing of the resource.
+ */
+@InternalApi
+public open class SdkManagedCloseable(private val closeable: Closeable) : SdkManagedBase() {
+    override fun unshare(): Boolean {
+        val shouldClose = super.unshare()
+        if (shouldClose) closeable.close()
+        return shouldClose
+    }
+}

runtime/io/common/src/aws/smithy/kotlin/runtime/io/SdkManagedGroup.kt

@@ -0,0 +1,32 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.smithy.kotlin.runtime.io
+
+import aws.smithy.kotlin.runtime.util.InternalApi
+
+/**
+ * A collection of opaque resources that implement [SdkManaged].
+ */
+@InternalApi
+public class SdkManagedGroup(
+    private val resources: MutableList<SdkManaged> = mutableListOf(),
+) {
+    public fun add(resource: SdkManaged) {
+        resource.share()
+        resources.add(resource)
+    }
+
+    public fun unshareAll() {
+        resources.forEach { it.unshare() }
+    }
+}
+
+/**
+ * Delegate extension to add a resource to the group if applicable.
+ */
+@InternalApi
+public fun SdkManagedGroup.addIfManaged(resource: Any) {
+    if (resource is SdkManaged) add(resource)
+}

runtime/io/common/test/aws/smithy/kotlin/runtime/io/SdkManagedCloseableTest.kt

@@ -0,0 +1,73 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.smithy.kotlin.runtime.io
+
+import kotlin.test.*
+
+class SdkManagedCloseableTest {
+    class MockCloseableImpl : Closeable {
+        var isClosed = false
+        var closeCount = 0
+        override fun close() {
+            isClosed = true
+            ++closeCount
+        }
+    }
+
+    @Test
+    fun testShareCount() {
+        val closeable = MockCloseableImpl()
+        val wrapped = SdkManagedCloseable(closeable)
+
+        wrapped.share()
+        wrapped.share()
+        wrapped.unshare()
+        assertFalse(closeable.isClosed)
+    }
+
+    @Test
+    fun testCloseNoShare() {
+        val closeable = MockCloseableImpl()
+        val wrapped = SdkManagedCloseable(closeable)
+
+        wrapped.unshare()
+        assertTrue(closeable.isClosed)
+    }
+
+    @Test
+    fun testCloseWithShare() {
+        val closeable = MockCloseableImpl()
+        val wrapped = SdkManagedCloseable(closeable)
+
+        wrapped.share()
+        wrapped.unshare()
+        assertTrue(closeable.isClosed)
+    }
+
+    @Test
+    fun testInnerCloseIdempotent() {
+        val closeable = MockCloseableImpl()
+        val wrapped = SdkManagedCloseable(closeable)
+
+        wrapped.share()
+        wrapped.unshare()
+        wrapped.unshare()
+        assertTrue(closeable.isClosed)
+        assertEquals(1, closeable.closeCount)
+    }
+
+    @Test
+    fun testShareClosedResource() {
+        val closeable = MockCloseableImpl()
+        val wrapped = SdkManagedCloseable(closeable)
+
+        wrapped.share()
+        wrapped.unshare()
+        val ex = assertFailsWith<IllegalStateException> {
+            wrapped.share()
+        }
+        assertEquals("caller attempted to share() a fully unshared object", ex.message)
+    }
+}