feat: add codegen wrappers for retries

gradle.properties

@@ -37,6 +37,7 @@ kotlinxBenchmarkVersion=0.3.1
 
 # serialization
 gsonVersion=2.8.5
+kamlVersion=0.36.0
 xmlpullVersion=1.1.3.1
 xpp3Version=1.1.6
 

runtime/protocol/http-client-engines/http-client-engine-ktor/jvm/src/aws/smithy/kotlin/runtime/http/engine/ktor/KtorUtils.kt

@@ -97,6 +97,7 @@ internal class KtorContentStream(private val channel: ByteReadChannel) : SdkByte
 // wrapper around a ByteReadChannel that implements the content as an SDK (streaming) HttpBody
 internal class KtorHttpBody(channel: ByteReadChannel) : HttpBody.Streaming() {
     private val source = KtorContentStream(channel)
+    override fun deepCopy(): HttpBody = throwSingleConsumptionException()
     override fun readFrom(): SdkByteReadChannel = source
 }
 

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/Headers.kt

@@ -4,8 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.http
 
-import aws.smithy.kotlin.runtime.http.util.StringValuesMap
-import aws.smithy.kotlin.runtime.http.util.StringValuesMapBuilder
+import aws.smithy.kotlin.runtime.http.util.*
 import aws.smithy.kotlin.runtime.http.util.StringValuesMapImpl
 
 /**
@@ -35,12 +34,13 @@ private object EmptyHeaders : Headers {
 /**
  * Build an immutable HTTP header map
  */
-class HeadersBuilder : StringValuesMapBuilder(true, 8) {
+class HeadersBuilder : StringValuesMapBuilder(true, 8), CanDeepCopy<HeadersBuilder> {
     override fun toString(): String = "HeadersBuilder ${entries()} "
-    override fun build(): Headers {
-        require(!built) { "HeadersBuilder can only build a single Headers instance" }
-        built = true
-        return HeadersImpl(values)
+    override fun build(): Headers = HeadersImpl(values)
+
+    override fun deepCopy(): HeadersBuilder {
+        val originalValues = values.deepCopy()
+        return HeadersBuilder().apply { values.putAll(originalValues) }
     }
 }
 

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/HttpBody.kt

@@ -5,24 +5,33 @@
 package aws.smithy.kotlin.runtime.http
 
 import aws.smithy.kotlin.runtime.content.ByteStream
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
 
 /**
  * HTTP payload to be sent to a peer
  */
-sealed class HttpBody {
+sealed class HttpBody : CanDeepCopy<HttpBody> {
+    /**
+     * Flag indicating the body can be consumed multiple times.
+     */
+    open val isReplayable: Boolean = true
 
     /**
      * Specifies the length of this [HttpBody] content
      * If null it is assumed to be a streaming source using e.g. `Transfer-Encoding: Chunked`
      */
     open val contentLength: Long? = null
 
+    abstract fun reset()
+
     /**
      * Variant of a [HttpBody] without a payload
      */
     object Empty : HttpBody() {
         override val contentLength: Long = 0
+        override fun deepCopy(): HttpBody = this // Deep copies are unnecessary for empty bodies
+        override fun reset() { } // Resets are unnecessary for empty bodies
     }
 
     /**
@@ -35,6 +44,14 @@ sealed class HttpBody {
          * Provides [ByteArray] to be sent to peer
          */
         abstract fun bytes(): ByteArray
+
+        override fun deepCopy(): Bytes = object : Bytes() {
+            private val copiedBytes = [email protected]().copyOf()
+            override fun bytes(): ByteArray = copiedBytes
+            override val contentLength: Long? = copiedBytes.size.toLong()
+        }
+
+        override fun reset() { } // Resets are unnecessary for byte bodies
     }
 
     /**
@@ -52,14 +69,20 @@ sealed class HttpBody {
          * Flag indicating the stream can be consumed multiple times. If `false` [reset] will throw an
          * [UnsupportedOperationException].
          */
-        open val isReplayable: Boolean = false
+        override val isReplayable: Boolean = false
 
         /**
          * Reset the stream such that the next call to [readFrom] provides a fresh channel.
          * @throws UnsupportedOperationException if the stream can only be consumed once. Consumers can check
          * [isReplayable] before calling
          */
-        open fun reset() { throw UnsupportedOperationException("${this::class.simpleName} can only be consumed once") }
+        open override fun reset() { throwSingleConsumptionException() }
+
+        /**
+         * Throw a general exception upon attempting to consume the stream more than once.
+         */
+        protected fun throwSingleConsumptionException(): Nothing =
+            throw UnsupportedOperationException("${this::class.simpleName} can only be consumed once")
     }
 }
 
@@ -73,11 +96,13 @@ fun ByteStream.toHttpBody(): HttpBody = when (val byteStream = this) {
     }
     is ByteStream.OneShotStream -> object : HttpBody.Streaming() {
         override val contentLength: Long? = byteStream.contentLength
+        override fun deepCopy(): Streaming = throwSingleConsumptionException()
         override fun readFrom(): SdkByteReadChannel = byteStream.readFrom()
     }
     is ByteStream.ReplayableStream -> object : HttpBody.Streaming() {
         private var channel: SdkByteReadChannel? = null
         override val contentLength: Long? = byteStream.contentLength
+        override fun deepCopy(): Streaming = this // Replayable streams copy themselves by default
         override fun readFrom(): SdkByteReadChannel = channel ?: byteStream.newReader().also { channel = it }
         override val isReplayable: Boolean = true
         override fun reset() {

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/QueryParameters.kt

@@ -4,9 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.http
 
-import aws.smithy.kotlin.runtime.http.util.StringValuesMap
-import aws.smithy.kotlin.runtime.http.util.StringValuesMapBuilder
-import aws.smithy.kotlin.runtime.http.util.StringValuesMapImpl
+import aws.smithy.kotlin.runtime.http.util.*
 import aws.smithy.kotlin.runtime.util.text.urlEncodeComponent
 
 /**
@@ -33,12 +31,13 @@ private object EmptyQueryParameters : QueryParameters {
     override fun isEmpty(): Boolean = true
 }
 
-class QueryParametersBuilder : StringValuesMapBuilder(true, 8) {
+class QueryParametersBuilder : StringValuesMapBuilder(true, 8), CanDeepCopy<QueryParametersBuilder> {
     override fun toString(): String = "QueryParametersBuilder ${entries()} "
-    override fun build(): QueryParameters {
-        require(!built) { "QueryParametersBuilder can only build a single instance" }
-        built = true
-        return QueryParametersImpl(values)
+    override fun build(): QueryParameters = QueryParametersImpl(values)
+
+    override fun deepCopy(): QueryParametersBuilder {
+        val originalValues = values.deepCopy()
+        return QueryParametersBuilder().apply { values.putAll(originalValues) }
     }
 }
 

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/Url.kt

@@ -4,6 +4,7 @@
  */
 package aws.smithy.kotlin.runtime.http
 
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 import aws.smithy.kotlin.runtime.util.InternalApi
 import aws.smithy.kotlin.runtime.util.text.encodeUrlPath
 
@@ -105,7 +106,7 @@ data class UserInfo(val username: String, val password: String)
 /**
  * Construct a URL by it's individual components
  */
-class UrlBuilder {
+class UrlBuilder : CanDeepCopy<UrlBuilder> {
     var scheme = Protocol.HTTPS
     var host: String = ""
     var port: Int? = null
@@ -130,6 +131,17 @@ class UrlBuilder {
         forceQuery
     )
 
+    override fun deepCopy(): UrlBuilder = UrlBuilder().apply {
+        scheme = [email protected]
+        host = [email protected]
+        port = [email protected]
+        path = [email protected]
+        parameters = [email protected]()
+        fragment = [email protected]
+        userInfo = [email protected]?.copy()
+        forceQuery = [email protected]
+    }
+
     override fun toString(): String =
         "UrlBuilder(scheme=$scheme, host='$host', port=$port, path='$path', parameters=$parameters, fragment=$fragment, userInfo=$userInfo, forceQuery=$forceQuery)"
 }

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/middleware/RetryFeature.kt

@@ -0,0 +1,58 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+package aws.smithy.kotlin.runtime.http.middleware
+
+import aws.smithy.kotlin.runtime.http.Feature
+import aws.smithy.kotlin.runtime.http.FeatureKey
+import aws.smithy.kotlin.runtime.http.HttpClientFeatureFactory
+import aws.smithy.kotlin.runtime.http.operation.SdkHttpOperation
+import aws.smithy.kotlin.runtime.http.operation.deepCopy
+import aws.smithy.kotlin.runtime.http.request.HttpRequestBuilder
+import aws.smithy.kotlin.runtime.retries.RetryPolicy
+import aws.smithy.kotlin.runtime.retries.RetryStrategy
+
+class RetryFeature(private val strategy: RetryStrategy, private val policy: RetryPolicy<Any?>) : Feature {
+    class Config {
+        var strategy: RetryStrategy? = null
+        var policy: RetryPolicy<Any?>? = null
+    }
+
+    companion object Feature : HttpClientFeatureFactory<Config, RetryFeature> {
+        override val key: FeatureKey<RetryFeature> = FeatureKey("RetryFeature")
+
+        override fun create(block: Config.() -> Unit): RetryFeature {
+            val config = Config().apply(block)
+            val strategy = requireNotNull(config.strategy) { "strategy is required" }
+            val policy = requireNotNull(config.policy) { "policy is required" }
+            return RetryFeature(strategy, policy)
+        }
+    }
+
+    override fun <I, O> install(operation: SdkHttpOperation<I, O>) {
+        operation.execution.finalize.intercept { req, next ->
+            if (req.subject.isRetryable) {
+                strategy.retry(policy) {
+                    // Deep copy the request because later middlewares (e.g., signing) mutate it
+                    val reqCopy = req.deepCopy()
+
+                    // Reset bodies back to beginning (mainly for streaming bodies)
+                    reqCopy.subject.body.reset()
+
+                    next.call(reqCopy)
+                }
+            } else {
+                next.call(req)
+            }
+        }
+    }
+}
+
+/**
+ * Indicates whether this HTTP request could be retried. Some requests with streaming bodies are unsuitable for
+ * retries.
+ */
+val HttpRequestBuilder.isRetryable: Boolean
+    get() = body.isReplayable

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/operation/OperationRequest.kt

@@ -6,6 +6,7 @@
 package aws.smithy.kotlin.runtime.http.operation
 
 import aws.smithy.kotlin.runtime.client.ExecutionContext
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 
 /**
  * Wrapper around a type [subject] with an execution context.
@@ -17,3 +18,11 @@ import aws.smithy.kotlin.runtime.client.ExecutionContext
  * @param subject The input type
  */
 data class OperationRequest<T>(val context: ExecutionContext, val subject: T)
+
+/**
+ * Deep copy an [OperationRequest] to a new request. Note that, because [context] is...well, context...it's considered
+ * transient and is not part of the copy. The subject itself, however, is deeply copied.
+ * @return A new [OperationRequest] with the same context and a deeply-copied subject.
+ */
+internal fun <T : CanDeepCopy<T>> OperationRequest<T>.deepCopy(): OperationRequest<T> =
+    OperationRequest(context, subject.deepCopy())

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/request/HttpRequestBuilder.kt

@@ -6,35 +6,30 @@ package aws.smithy.kotlin.runtime.http.request
 
 import aws.smithy.kotlin.runtime.http.*
 import aws.smithy.kotlin.runtime.http.content.ByteArrayContent
+import aws.smithy.kotlin.runtime.http.util.CanDeepCopy
 import aws.smithy.kotlin.runtime.io.*
 import aws.smithy.kotlin.runtime.util.InternalApi
 
 /**
  * Used to construct an HTTP request
+ * @param method The HTTP method (verb) to use when making the request
+ * @param url Endpoint to make request to
+ * @param headers HTTP headers
+ * @param body Outgoing payload. Initially empty
  */
-class HttpRequestBuilder {
-    /**
-     * The HTTP method (verb) to use when making the request
-     */
-    var method: HttpMethod = HttpMethod.GET
-
-    /**
-     * Endpoint to make request to
-     */
-    val url: UrlBuilder = UrlBuilder()
-
-    /**
-     * HTTP headers
-     */
-    val headers: HeadersBuilder = HeadersBuilder()
-
-    /**
-     * Outgoing payload. Initially empty
-     */
-    var body: HttpBody = HttpBody.Empty
+class HttpRequestBuilder private constructor(
+    var method: HttpMethod,
+    val url: UrlBuilder,
+    val headers: HeadersBuilder,
+    var body: HttpBody,
+) : CanDeepCopy<HttpRequestBuilder> {
+    constructor() : this(HttpMethod.GET, UrlBuilder(), HeadersBuilder(), HttpBody.Empty)
 
     fun build(): HttpRequest = HttpRequest(method, url.build(), if (headers.isEmpty()) Headers.Empty else headers.build(), body)
 
+    override fun deepCopy(): HttpRequestBuilder =
+        HttpRequestBuilder(method, url.deepCopy(), headers.deepCopy(), body.deepCopy())
+
     override fun toString(): String = buildString {
         append("HttpRequestBuilder(method=$method, url=$url, headers=$headers, body=$body)")
     }

runtime/protocol/http/common/src/aws/smithy/kotlin/runtime/http/util/CanDeepCopy.kt

@@ -0,0 +1,10 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+package aws.smithy.kotlin.runtime.http.util
+
+interface CanDeepCopy<T> {
+    fun deepCopy(): T
+}