Merge request and response types

Sources/SotoCore/AWSClient.swift

@@ -121,7 +121,7 @@
             do {
                 try await shutdown()
             } catch {
                 errorStorage.withLockedValue { errorStorage in
                     errorStorage = error
                 }
             }
@@ -204,7 +204,7 @@
     /// the HTTP client if it was created by the `AWSClient`.
     public func shutdown() async throws {
         guard self.isShutdown.compareExchange(expected: false, desired: true, ordering: .relaxed).exchanged else {
             throw ClientError.alreadyShutdown
         }
         // shutdown credential provider ignoring any errors as credential provider that doesn't initialize
         // can cause the shutdown process to fail
@@ -215,10 +215,10 @@
             do {
                 try await self.httpClient.shutdown()
             } catch {
                 self.clientLogger.log(level: self.options.errorLogLevel, "Error shutting down HTTP client", metadata: [
                     "aws-error": "\(error)",
                 ])
                 throw error
             }
 
         case .shared:
@@ -247,18 +247,22 @@
         return try await self.execute(
             operation: operationName,
             createRequest: {
-                try AWSRequest(
+                try AWSHTTPRequest(
                     operation: operationName,
                     path: path,
-                    httpMethod: httpMethod,
+                    method: httpMethod,
                     input: input,
                     hostPrefix: hostPrefix,
                     configuration: serviceConfig
                 )
             },
             processResponse: { response in
-                // flush response body contents to complete response read
-                for try await _ in response.body {}
+                return try await self.processEmptyResponse(
+                    operation: operationName,
+                    response: response,
+                    serviceConfig: serviceConfig,
+                    logger: logger
+                )
             },
             config: serviceConfig,
             logger: logger
@@ -282,16 +286,20 @@
         return try await self.execute(
             operation: operationName,
             createRequest: {
-                try AWSRequest(
+                try AWSHTTPRequest(
                     operation: operationName,
                     path: path,
-                    httpMethod: httpMethod,
+                    method: httpMethod,
                     configuration: serviceConfig
                 )
             },
             processResponse: { response in
-                // flush response body contents to complete response read
-                for try await _ in response.body {}
+                return try await self.processEmptyResponse(
+                    operation: operationName,
+                    response: response,
+                    serviceConfig: serviceConfig,
+                    logger: logger
+                )
             },
             config: serviceConfig,
             logger: logger
@@ -317,15 +325,20 @@
         return try await self.execute(
             operation: operationName,
             createRequest: {
-                try AWSRequest(
+                try AWSHTTPRequest(
                     operation: operationName,
                     path: path,
-                    httpMethod: httpMethod,
+                    method: httpMethod,
                     configuration: serviceConfig
                 )
             },
             processResponse: { response in
-                return try await self.validate(operation: operationName, response: response, serviceConfig: serviceConfig)
+                return try await self.processResponse(
+                    operation: operationName,
+                    response: response,
+                    serviceConfig: serviceConfig,
+                    logger: logger
+                )
             },
             config: serviceConfig,
             logger: logger
@@ -355,17 +368,17 @@
         return try await self.execute(
             operation: operationName,
             createRequest: {
-                try AWSRequest(
+                try AWSHTTPRequest(
                     operation: operationName,
                     path: path,
-                    httpMethod: httpMethod,
+                    method: httpMethod,
                     input: input,
                     hostPrefix: hostPrefix,
                     configuration: serviceConfig
                 )
             },
             processResponse: { response in
-                return try await self.validate(operation: operationName, response: response, serviceConfig: serviceConfig)
+                return try await self.processResponse(operation: operationName, response: response, serviceConfig: serviceConfig, logger: logger)
             },
             config: serviceConfig,
             logger: logger
@@ -375,7 +388,7 @@
     /// internal version of execute
     internal func execute<Output>(
         operation operationName: String,
-        createRequest: @escaping () throws -> AWSRequest,
+        createRequest: @escaping () throws -> AWSHTTPRequest,
         processResponse: @escaping (AWSHTTPResponse) async throws -> Output,
         config: AWSServiceConfig,
         logger: Logger = AWSClient.loggingDisabled
@@ -396,18 +409,17 @@
             // construct signer
             let signer = AWSSigner(credentials: credential, name: config.signingName, region: config.region.rawValue)
             // create request and sign with signer
-            let awsRequest = try createRequest()
-                .applyMiddlewares(config.middlewares + self.middlewares, config: config)
-                .createHTTPRequest(signer: signer, serviceConfig: config)
-            // send request to AWS and process result
-            let streaming = awsRequest.body.isStreaming
+            var request = try createRequest()
+                .applyMiddlewares(config.middlewares + self.middlewares, context: .init(operation: operationName, serviceConfig: config))
+            request.signHeaders(signer: signer, serviceConfig: config)
             try Task.checkCancellation()
+            // apply middleware and sign
             let response = try await self.invoke(
-                request: awsRequest,
+                request: request,
+                operation: operationName,
                 with: config,
                 logger: logger,
-                processResponse: processResponse,
-                streaming: streaming
+                processResponse: processResponse
             )
             logger.trace("AWS Response")
             Metrics.Timer(
@@ -431,30 +443,25 @@
 
     func invoke<Output>(
         request: AWSHTTPRequest,
+        operation operationName: String,
         with serviceConfig: AWSServiceConfig,
         logger: Logger,
-        processResponse: @escaping (AWSHTTPResponse) async throws -> Output,
-        streaming: Bool
+        processResponse: @escaping (AWSHTTPResponse) async throws -> Output
     ) async throws -> Output {
         var attempt = 0
         while true {
             do {
                 let response = try await self.httpClient.execute(request: request, timeout: serviceConfig.timeout, logger: logger)
-                // if it returns an HTTP status code outside 2xx then throw an error
+                    .applyMiddlewares(serviceConfig.middlewares + self.middlewares, context: .init(operation: operationName, serviceConfig: serviceConfig))
+                // if response has an HTTP status code outside 2xx then throw an error
                 guard (200..<300).contains(response.status.code) else {
-                    let error = try await self.createError(for: response, serviceConfig: serviceConfig, logger: logger)
+                    let error = await self.createError(for: response, serviceConfig: serviceConfig, logger: logger)
                     throw error
                 }
+
                 let output = try await processResponse(response)
                 return output
             } catch {
-                // if streaming and the error returned is an AWS error fail immediately. Do not attempt
-                // to retry as the streaming function will not know you are retrying
-                if streaming,
-                   error is AWSErrorType || error is AWSRawError
-                {
-                    throw error
-                }
                 // If I get a retry wait time for this error then attempt to retry request
                 if case .retry(let retryTime) = self.retryPolicy.getRetryWaitTime(error: error, attempt: attempt) {
                     logger.trace("Retrying request", metadata: [
@@ -502,7 +509,7 @@
         )
         let signer = try await self.createSigner(serviceConfig: serviceConfig, logger: logger)
         guard let cleanURL = signer.processURL(url: url) else {
             throw AWSClient.ClientError.invalidURL
         }
         return signer.signURL(url: cleanURL, method: httpMethod, headers: headers, expires: expires)
     }
@@ -532,14 +539,14 @@
         )
         let signer = try await self.createSigner(serviceConfig: serviceConfig, logger: logger)
         guard let cleanURL = signer.processURL(url: url) else {
             throw AWSClient.ClientError.invalidURL
         }
         let bodyData: AWSSigner.BodyData?
         switch body.storage {
         case .byteBuffer(let buffer):
             bodyData = .byteBuffer(buffer)
         case .asyncSequence:
             bodyData = nil
         }
         return signer.signHeaders(url: cleanURL, method: httpMethod, headers: headers, body: bodyData)
     }
@@ -553,61 +560,67 @@
 // response validator
 extension AWSClient {
     /// Generate an AWS Response from  the operation HTTP response and return the output shape from it. This is only every called if the response includes a successful http status code
-    internal func validate<Output: AWSDecodableShape>(
+    internal func processResponse<Output: AWSDecodableShape>(
         operation operationName: String,
         response: AWSHTTPResponse,
-        serviceConfig: AWSServiceConfig
+        serviceConfig: AWSServiceConfig,
+        logger: Logger
     ) async throws -> Output {
-        assert((200..<300).contains(response.status.code), "Shouldn't get here if error was returned")
-
         let raw = Output._options.contains(.rawPayload) == true
-        let awsResponse = try await AWSResponse(from: response, streaming: raw)
-            .applyMiddlewares(serviceConfig.middlewares + middlewares, config: serviceConfig)
+        var response = response
+        if !raw {
+            try await response.collateBody()
+        }
+        return try response.generateOutputShape(operation: operationName, serviceProtocol: serviceConfig.serviceProtocol)
+    }
 
-        return try awsResponse.generateOutputShape(operation: operationName, serviceProtocol: serviceConfig.serviceProtocol)
+    /// Generate an AWS Response from the operation HTTP response and return the output shape from it. This is only ever called if the response includes a successful http status code
+    internal func processEmptyResponse(
+        operation operationName: String,
+        response: AWSHTTPResponse,
+        serviceConfig: AWSServiceConfig,
+        logger: Logger
+    ) async throws {
+        // flush response body contents to complete response read
+        for try await _ in response.body {}
     }
 
     /// Create error from HTTPResponse. This is only called if we received an unsuccessful http status code.
-    internal func createError(for response: AWSHTTPResponse, serviceConfig: AWSServiceConfig, logger: Logger) async throws -> Error {
+    internal func createError(for response: AWSHTTPResponse, serviceConfig: AWSServiceConfig, logger: Logger) async -> Error {
         // if we can create an AWSResponse and create an error from it return that
-        let awsResponse: AWSResponse
+        var response = response
         do {
-            awsResponse = try await AWSResponse(from: response, streaming: false)
+            try await response.collateBody()
         } catch {
             // else return "Unhandled error message" with rawBody attached
             let context = AWSErrorContext(
                 message: "Unhandled Error",
                 responseCode: response.status,
                 headers: response.headers
             )
             return AWSRawError(rawBody: nil, context: context)
         }
-        do {
-            let awsResponseWithMiddleware = try awsResponse.applyMiddlewares(serviceConfig.middlewares + middlewares, config: serviceConfig)
-            if let error = awsResponseWithMiddleware.generateError(
-                serviceConfig: serviceConfig,
-                logLevel: options.errorLogLevel,
-                logger: logger
-            ) {
-                return error
-            } else {
-                // else return "Unhandled error message" with rawBody attached
-                let context = AWSErrorContext(
-                    message: "Unhandled Error",
-                    responseCode: response.status,
-                    headers: response.headers
-                )
-                let responseBody: String?
-                switch awsResponseWithMiddleware.body.storage {
-                case .byteBuffer(let buffer):
-                    responseBody = String(buffer: buffer)
-                default:
-                    responseBody = nil
-                }
-                return AWSRawError(rawBody: responseBody, context: context)
-            }
-        } catch {
+        if let error = response.generateError(
+            serviceConfig: serviceConfig,
+            logLevel: options.errorLogLevel,
+            logger: logger
+        ) {
             return error
+        } else {
+            // else return "Unhandled error message" with rawBody attached
+            let context = AWSErrorContext(
+                message: "Unhandled Error",
+                responseCode: response.status,
+                headers: response.headers
+            )
+            let responseBody: String?
+            switch response.body.storage {
+            case .byteBuffer(let buffer):
+                responseBody = String(buffer: buffer)
+            default:
+                responseBody = nil
+            }
+            return AWSRawError(rawBody: responseBody, context: context)
         }
     }
 }

Sources/SotoCore/Encoder/QueryEncoder.swift

@@ -55,7 +55,7 @@ public struct QueryEncoder {
     static let queryAllowedCharacters = CharacterSet(charactersIn: "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~")
 
     private static func urlEncodeQueryParam(_ value: String) -> String {
-        return value.addingPercentEncoding(withAllowedCharacters: AWSRequest.queryAllowedCharacters) ?? value
+        return value.addingPercentEncoding(withAllowedCharacters: AWSHTTPRequest.queryAllowedCharacters) ?? value
     }
 
     // generate string from

Sources/SotoCore/Encoder/ResponseContainer.swift

@@ -21,34 +21,34 @@
     let header: String
     let message: String
 
     static func headerNotFound(_ header: String) -> Self { .init(header: header, message: "Header not found") }
     static func typeMismatch(_ header: String, expectedType: String) -> Self { .init(header: header, message: "Cannot convert header to \(expectedType)") }
 }
 
 public struct ResponseDecodingContainer {
-    let response: AWSResponse
+    let response: AWSHTTPResponse
 
     public func decode<Value: RawRepresentable>(_ type: Value.Type = Value.self, forHeader header: String) throws -> Value where Value.RawValue == String {
         guard let headerValue = response.headers[header].first else {
             throw HeaderDecodingError.headerNotFound(header)
         }
         if let result = Value(rawValue: headerValue) {
             return result
         } else {
             throw HeaderDecodingError.typeMismatch(header, expectedType: "\(Value.self)")
         }
     }
 
     public func decode<Value: LosslessStringConvertible>(_ type: Value.Type = Value.self, forHeader header: String) throws -> Value {
         guard let headerValue = response.headers[header].first else {
             throw HeaderDecodingError.headerNotFound(header)
         }
         if let result = Value(headerValue) {
             return result
         } else {
             throw HeaderDecodingError.typeMismatch(header, expectedType: "\(Value.self)")
         }
     }
 
     public func decodeStatus<Value: FixedWidthInteger>(_: Value.Type = Value.self) -> Value {
         return Value(self.response.status.code)
@@ -58,13 +58,13 @@
         return self.response.body
     }
 
     public func decodeEventStream<Event>() -> AWSEventStream<Event> {
         return .init(self.response.body)
     }
 
     public func decode(_ type: Date.Type = Date.self, forHeader header: String) throws -> Date {
         guard let headerValue = response.headers[header].first else {
             throw HeaderDecodingError.headerNotFound(header)
         }
         let dateFormatter = DateFormatter()
         dateFormatter.locale = Locale(identifier: "en_US_POSIX")
@@ -73,45 +73,45 @@
         if let result = dateFormatter.date(from: headerValue) {
             return result
         } else {
             throw HeaderDecodingError.typeMismatch(header, expectedType: "Date")
         }
     }
 
     public func decodeIfPresent<Value: RawRepresentable>(_ type: Value.Type = Value.self, forHeader header: String) throws -> Value? where Value.RawValue == String {
         guard let headerValue = response.headers[header].first else { return nil }
         if let result = Value(rawValue: headerValue) {
             return result
         } else {
             throw HeaderDecodingError.typeMismatch(header, expectedType: "\(Value.self)")
         }
     }
 
     public func decodeIfPresent<Value: LosslessStringConvertible>(_ type: Value.Type = Value.self, forHeader header: String) throws -> Value? {
         guard let headerValue = response.headers[header].first else { return nil }
         if let result = Value(headerValue) {
             return result
         } else {
             throw HeaderDecodingError.typeMismatch(header, expectedType: "\(Value.self)")
         }
     }
 
     public func decodeIfPresent(_ type: Date.Type = Date.self, forHeader header: String) throws -> Date? {
         guard let headerValue = response.headers[header].first else { return nil }
         let dateFormatter = DateFormatter()
         dateFormatter.locale = Locale(identifier: "en_US_POSIX")
         dateFormatter.dateFormat = "EEE, d MMM yyy HH:mm:ss z"
         dateFormatter.timeZone = TimeZone(secondsFromGMT: 0)
         if let result = dateFormatter.date(from: headerValue) {
             return result
         } else {
             throw HeaderDecodingError.typeMismatch(header, expectedType: "Date")
         }
     }
 
     public func decodeIfPresent(_ type: [String: String].Type = [String: String].self, forHeader header: String) throws -> [String: String]? {
         let headers = self.response.headers.compactMap { $0.name.hasPrefix(header) ? $0 : nil }
         if headers.count == 0 {
             return nil
         }
         return [String: String](headers.map { (key: String($0.name.dropFirst(header.count)), value: $0.value) }) { first, _ in first }
     }

Sources/SotoCore/HTTP/AWSHTTPTypes.swift → Sources/SotoCore/HTTP/AWSHTTPBody.swift

@@ -54,29 +54,29 @@
     public func collect(upTo length: Int) async throws -> ByteBuffer {
         switch self.storage {
         case .byteBuffer(let buffer):
             return buffer
         case .asyncSequence(let sequence, _):
             return try await sequence.collect(upTo: length)
         }
     }
 
     public var length: Int? {
         switch self.storage {
         case .byteBuffer(let buffer):
             return buffer.readableBytes
         case .asyncSequence(_, let length):
             return length
         }
     }
 
     public var isStreaming: Bool {
         switch self.storage {
         case .byteBuffer:
             return false
         case .asyncSequence:
             return true
         }
     }
 }
 
 extension AWSHTTPBody: AsyncSequence {
@@ -86,7 +86,7 @@
     public func makeAsyncIterator() -> AsyncIterator {
         switch self.storage {
         case .byteBuffer(let buffer):
             return AnyAsyncSequence(buffer.asyncSequence(chunkSize: buffer.readableBytes)).makeAsyncIterator()
         case .asyncSequence(let sequence, _):
             return sequence.makeAsyncIterator()
         }
@@ -96,41 +96,7 @@
 extension AWSHTTPBody: Decodable {
     // AWSHTTPBody has to conform to Decodable so I can add it to AWSShape objects (which conform to Decodable). But we don't want the
     // Encoder/Decoder ever to process a AWSPayload
     public init(from decoder: Decoder) throws {
         preconditionFailure("Cannot decode an AWSHTTPBody")
     }
 }
-
-/// HTTP Request
-struct AWSHTTPRequest {
-    let url: URL
-    let method: HTTPMethod
-    let headers: HTTPHeaders
-    let body: AWSHTTPBody
-
-    init(url: URL, method: HTTPMethod, headers: HTTPHeaders = [:], body: AWSHTTPBody = .init()) {
-        self.url = url
-        self.method = method
-        self.headers = headers
-        self.body = body
-    }
-}
-
-/// Generic HTTP Response returned from HTTP Client
-struct AWSHTTPResponse: Sendable {
-    /// Initialize AWSHTTPResponse
-    init(status: HTTPResponseStatus, headers: HTTPHeaders, body: AWSHTTPBody = .init()) {
-        self.status = status
-        self.headers = headers
-        self.body = body
-    }
-
-    /// The HTTP status for this response.
-    var status: HTTPResponseStatus
-
-    /// The HTTP headers of this response.
-    var headers: HTTPHeaders
-
-    /// The body of this HTTP response.
-    var body: AWSHTTPBody
-}