fix: Streaming round trip (#284)

Author: kneekey23

Packages/ClientRuntime/Sources/Networking/Http/CRT/CRTClientEngine.swift

@@ -79,28 +79,6 @@ public class CRTClientEngine: HttpClientEngine {
         return connectionPool
     }
 
-    private func addHttpHeaders(endpoint: Endpoint, request: SdkHttpRequest) -> HttpRequest {
-        
-        var headers = request.headers
-        
-        let contentLength: Int64 = {
-            switch request.body {
-            case .data(let data):
-                return Int64(data?.count ?? 0)
-            case .streamSource(let stream):
-                // TODO: implement dynamic streaming with transfer-encoded-chunk header
-                return stream.unwrap().contentLength
-            case .none, .streamSink:
-                return 0
-            }
-        }()
-        
-        headers.update(name: CONTENT_LENGTH_HEADER, value: "\(contentLength)")
-        
-        request.headers = headers
-        return request.toHttpRequest(bufferSize: windowSize)
-    }
-    
     public func executeWithClosure(request: SdkHttpRequest, completion: @escaping NetworkResult) {
         execute(request: request).then { (result) in
             completion(result)
@@ -110,7 +88,7 @@ public class CRTClientEngine: HttpClientEngine {
     public func execute(request: SdkHttpRequest) -> Future<HttpResponse> {
         let isStreaming = { () -> Bool in
             switch request.body {
-            case .streamSink, .streamSource: return true
+            case .stream: return true
             default: return false
             }
         }()
@@ -151,15 +129,16 @@ public class CRTClientEngine: HttpClientEngine {
 
     public func makeHttpRequestStreamOptions(_ request: SdkHttpRequest) -> (HttpRequestOptions, Future<HttpResponse>) {
         let future = Future<HttpResponse>()
-        let requestWithHeaders = addHttpHeaders(endpoint: request.endpoint, request: request)
+        let crtRequest = request.toHttpRequest(bufferSize: windowSize)
         let response = HttpResponse()
 
-        var streamSink: StreamSink?
-        if case let HttpBody.streamSink(unwrappedStream) = request.body {
-            // we know they want to receive a stream via their request body type
-            streamSink = unwrappedStream.unwrap()
+        var streamReader: StreamReader?
+        if case let HttpBody.stream(unwrappedStream) = request.body,
+           case let ByteStream.reader(reader) = unwrappedStream {
+            streamReader = reader
         }
-        let requestOptions = HttpRequestOptions(request: requestWithHeaders) { [self] (stream, _, httpHeaders) in
+        
+        let requestOptions = HttpRequestOptions(request: crtRequest) { [self] (stream, _, httpHeaders) in
             logger.debug("headers were received")
             response.statusCode = HttpStatusCode(rawValue: Int(stream.getResponseStatusCode()))
                 ?? HttpStatusCode.notFound
@@ -171,28 +150,25 @@ public class CRTClientEngine: HttpClientEngine {
         } onIncomingBody: { [self] (_, data) in
             logger.debug("incoming data")
 
-            if let streamSink = streamSink {
+            if let streamReader = streamReader {
                 let byteBuffer = ByteBuffer(data: data)
-                streamSink.receiveData(readFrom: byteBuffer)
+                streamReader.write(buffer: byteBuffer)
             }
         } onStreamComplete: { [self] (_, error) in
             logger.debug("stream completed")
             if case let CRTError.crtError(unwrappedError) = error {
                 if unwrappedError.errorCode != 0 {
                     logger.error("Response encountered an error: \(error)")
-                    if let streamSink = streamSink {
-                        streamSink.onError(error: StreamError.unknown(error))
+                    if let streamReader = streamReader {
+                        streamReader.onError(error: ClientError.crtError(error))
                     }
                     future.fail(error)
                 }
             }
-            
-            if let streamSink = streamSink {
-                response.body = HttpBody.streamSink(.provider(streamSink))
-            } else {
-                response.body = HttpBody.none
+            if let streamReader = streamReader {
+                streamReader.hasFinishedWriting = true
+                response.body = .stream(.reader(streamReader))
             }
-            
             future.fulfill(response)
         }
 
@@ -201,12 +177,12 @@ public class CRTClientEngine: HttpClientEngine {
 
     public func makeHttpRequestOptions(_ request: SdkHttpRequest) -> (HttpRequestOptions, Future<HttpResponse>) {
         let future = Future<HttpResponse>()
-        let requestWithHeaders = addHttpHeaders(endpoint: request.endpoint, request: request)
+        let crtRequest = request.toHttpRequest(bufferSize: windowSize)
 
         let response = HttpResponse()
-        let incomingByteBuffer = ByteBuffer(size: 0)
+        var incomingData = Data()
 
-        let requestOptions = HttpRequestOptions(request: requestWithHeaders) { [self] (stream, _, httpHeaders) in
+        let requestOptions = HttpRequestOptions(request: crtRequest) { [self] (stream, _, httpHeaders) in
             logger.debug("headers were received")
             response.statusCode = HttpStatusCode(rawValue: Int(stream.getResponseStatusCode()))
                 ?? HttpStatusCode.notFound
@@ -216,8 +192,8 @@ public class CRTClientEngine: HttpClientEngine {
             response.statusCode = HttpStatusCode(rawValue: Int(stream.getResponseStatusCode()))
                 ?? HttpStatusCode.notFound
         } onIncomingBody: { [self] (_, data) in
-            logger.debug("incoming data")
-            incomingByteBuffer.put(data)
+            logger.debug("incoming data: \(data.count) bytes")
+            incomingData.append(data)
         } onStreamComplete: { [self] (_, error) in
             logger.debug("stream completed")
             if case let CRTError.crtError(unwrappedError) = error {
@@ -227,7 +203,7 @@ public class CRTClientEngine: HttpClientEngine {
                 }
             }
 
-            response.body = HttpBody.data(incomingByteBuffer.toData())
+            response.body = HttpBody.data(incomingData)
             future.fulfill(response)
         }
 

Packages/ClientRuntime/Sources/Networking/Http/DefaultMiddlewares/ContentLengthMiddleware.swift

@@ -13,24 +13,24 @@ public struct ContentLengthMiddleware<OperationStackOutput: HttpResponseBinding,
                           input: MInput,
                           next: H) -> Result<MOutput, MError>
     where H: Handler,
-          Self.MInput == H.Input,
-          Self.MOutput == H.Output,
-          Self.Context == H.Context,
-          Self.MError == H.MiddlewareError {
+    Self.MInput == H.Input,
+    Self.MOutput == H.Output,
+    Self.Context == H.Context,
+    Self.MError == H.MiddlewareError {
 
-        let contentLength: Int64 = {
-            switch input.body {
-            case .data(let data):
-                return Int64(data?.count ?? 0)
-            case .streamSource(let stream):
-                // TODO: implement dynamic streaming with transfer-encoded-chunk header
-                return stream.unwrap().contentLength
-            case .none, .streamSink:
-                return 0
+        switch input.body {
+        case .data(let data):
+            input.headers.update(name: "Content-Length", value: String(data?.count ?? 0))
+        case .stream(let stream):
+            switch stream {
+            case .buffer(let bytes):
+                input.headers.update(name: "Content-Length", value: String(bytes.length))
+            case .reader(_):
+                input.headers.update(name: "Transfer-Encoded", value: "Chunked")
             }
-        }()
-        
-        input.headers.update(name: "Content-Length", value: String(contentLength))
+        default:
+            input.headers.update(name: "Content-Length", value: "0")
+        }
 
         return next.handle(context: context, input: input)
     }

Packages/ClientRuntime/Sources/Networking/Http/HttpBody.swift

@@ -4,36 +4,39 @@
  */
 import AwsCommonRuntimeKit
 
-public enum HttpBody {
+public enum HttpBody: Equatable {
     case data(Data?)
-    case streamSource(StreamSourceProvider)
-    case streamSink(StreamSinkProvider)
+    case stream(ByteStream)
     case none
 }
 
-extension HttpBody: Equatable {
-    public static func == (lhs: HttpBody, rhs: HttpBody) -> Bool {
-        switch (lhs, rhs) {
-        case (let .data(unwrappedlhsData), let .data(unwrappedRhsData)):
-            return unwrappedlhsData == unwrappedRhsData
-        case (.streamSource, .streamSource):
-            return false
-        case (.streamSink, .streamSink):
-            return false
-        case (.none, .none):
-            return true
-        default:
-            return false
-        }
-    }
-}
-
 public extension HttpBody {
     static var empty: HttpBody {
         .data(nil)
     }
 }
 
+extension HttpBody {
+    func toAwsInputStream() -> AwsInputStream? {
+        switch self {
+        case .data(let data):
+            guard let data = data else {
+                return nil
+            }
+            return AwsInputStream(ByteBuffer(data: data))
+        case .stream(let stream):
+            switch stream {
+            case .reader(let reader):
+                return AwsInputStream(reader.read(maxBytes: nil))
+            case .buffer(let byteBuffer):
+                return AwsInputStream(byteBuffer)
+            }
+        case .none:
+            return nil
+        }
+    }
+}
+
 extension HttpBody: CustomDebugStringConvertible {
     public var debugDescription: String {
         var bodyAsString: String?
@@ -42,10 +45,8 @@ extension HttpBody: CustomDebugStringConvertible {
             if let data = data {
                 bodyAsString = String(data: data, encoding: .utf8)
             }
-        case .streamSource(let stream):
-            let byteBuffer = ByteBuffer(size: 1024)
-            stream.unwrap().sendData(writeTo: byteBuffer)
-            bodyAsString = String(data: byteBuffer.toData(), encoding: .utf8)
+        case .stream(let stream):
+            bodyAsString = String(data: stream.toBytes().toData(), encoding: .utf8)
         default:
             bodyAsString = nil
         }

Packages/ClientRuntime/Sources/Networking/Http/SdkHttpRequest.swift

@@ -36,24 +36,7 @@ extension SdkHttpRequest {
         httpRequest.method = method.rawValue
         httpRequest.path = "\(endpoint.path)\(endpoint.queryItemString)"
         httpRequest.addHeaders(headers: httpHeaders)
-        var awsInputStream: AwsInputStream?
-        switch body {
-        case .data(let data):
-            if let data = data {
-                let byteBuffer = ByteBuffer(data: data)
-                awsInputStream = AwsInputStream(byteBuffer)
-            }
-        case .streamSource(let stream):
-            let byteBuffer = ByteBuffer(size: bufferSize)
-            stream.unwrap().sendData(writeTo: byteBuffer)
-            awsInputStream = AwsInputStream(byteBuffer)
-        case .none, .streamSink:
-            awsInputStream = nil
-        }
-        if let inputStream = awsInputStream {
-            httpRequest.body = inputStream
-        }
-        
+        httpRequest.body = body.toAwsInputStream()
         return httpRequest
     }
 }