swift-http-server: Use AsyncReader in RequestBody

Benchmarks/HummingbirdBenchmarks/RouterBenchmarks.swift

@@ -6,6 +6,7 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
+import BasicContainers
 import Benchmark
 import HTTPTypes
 import Hummingbird
@@ -65,7 +66,10 @@ extension Benchmark {
     ) where ResponderBuilder.Responder.Context: RequestContext, ResponderBuilder.Responder.Context.Source == BenchmarkRequestContextSource {
         let responder = createRouter().buildResponder()
 
-        let (requestBody, source) = RequestBody.makeStream()
+        let (stream, source) = NIOAsyncChannelInboundStream<HTTPRequestPart>.makeTestingStream()
+        let iterator = stream.makeAsyncIterator()
+        let reader = BaseRequestAsyncReader(readerState: .init(iterator: iterator))
+        let requestBody = RequestBody(.asyncReader(reader))
         let hbRequest = Request(head: request, body: requestBody)
         source.finish()
 
@@ -78,9 +82,12 @@ extension Benchmark {
 
                 for _ in benchmark.scaledIterations {
                     for _ in 0..<50 {
-                        let (requestBody, source) = RequestBody.makeStream()
+                        let (stream, source) = NIOAsyncChannelInboundStream<HTTPRequestPart>.makeTestingStream()
+                        let iterator = stream.makeAsyncIterator()
+                        let reader = BaseRequestAsyncReader(readerState: .init(iterator: iterator))
+                        let requestBody = RequestBody(.asyncReader(reader))
                         let request = Request(head: request, body: requestBody)
-                        try await writeBody(source.yield)
+                        try await writeBody { source.yield(.body($0)) }
                         source.finish()
                         let response = try await responder.respond(to: request, context: context)
                         _ = try await response.body.write(BenchmarkBodyWriter())
@@ -153,11 +160,12 @@ func routerBenchmarks() {
         let router = Router(context: BasicBenchmarkContext.self)
         router.put { request, _ in
             let body = try await request.body.collect(upTo: .max)
-            return body.readableBytes.description
+            return body.count.description
         }
         return router
     }
 
+    /* TODO: Fixup for RequestAsyncReader
     Benchmark(
         "Router:Echo",
         configuration: .init(warmupIterations: 10),
@@ -183,7 +191,7 @@ func routerBenchmarks() {
         }
         return router
     }
-
+    */
     Benchmark(
         "Router:CaseInsensitive",
         configuration: .init(warmupIterations: 10),

Sources/Hummingbird/Codable/JSON/JSONCoding.swift

@@ -6,6 +6,7 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
+import BasicContainers
 private import NIOFoundationEssentialsCompat
 
 #if canImport(FoundationEssentials)
@@ -42,6 +43,9 @@ extension JSONDecoder: RequestDecoder {
     ///   - context: Request context
     public func decode<T: Decodable>(_ type: T.Type, from request: Request, context: some RequestContext) async throws -> T {
         let buffer = try await request.body.collect(upTo: context.maxUploadSize)
-        return try self.decode(T.self, from: buffer)
+        return try buffer.span.withUnsafeBytes { bytes in
+            let data = Data(bytes)
+            return try self.decode(T.self, from: data)
+        }
     }
 }

Sources/Hummingbird/Codable/URLEncodedForm/URLEncodedForm+Request.swift

@@ -5,6 +5,7 @@
 // See LICENSE.txt for license information
 // SPDX-License-Identifier: Apache-2.0
 //
+import BasicContainers
 
 @available(hummingbird 2.0, *)
 extension URLEncodedFormEncoder: ResponseEncoder {
@@ -36,7 +37,9 @@ extension URLEncodedFormDecoder: RequestDecoder {
     ///   - context: Request context
     public func decode<T: Decodable>(_ type: T.Type, from request: Request, context: some RequestContext) async throws -> T {
         let buffer = try await request.body.collect(upTo: context.maxUploadSize)
-        let string = String(buffer: buffer)
-        return try self.decode(T.self, from: string)
+        return try buffer.span.withUnsafeBytes { bytes in
+            let string = String(decoding: bytes, as: UTF8.self)
+            return try self.decode(T.self, from: string)
+        }
     }
 }

Sources/Hummingbird/Error/RequestAsyncReaderError+HTTPError.swift

@@ -0,0 +1,23 @@
+//
+// This source file is part of the Hummingbird server framework project
+// Copyright (c) the Hummingbird authors
+//
+// See LICENSE.txt for license information
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import HummingbirdCore
+
+// If we catch a too many bytes error report that as payload too large
+extension RequestAsyncReaderError: HTTPResponseError {
+    package var status: HTTPTypes.HTTPResponse.Status {
+        switch self {
+        case .streamEndedBeforeReceivingRequestEnd: .badRequest
+        case .tooLarge: .contentTooLarge
+        }
+    }
+
+    public func response(from request: Request, context: some RequestContext) throws -> Response {
+        Response(status: self.status)
+    }
+}

Sources/Hummingbird/Exports.swift

@@ -11,13 +11,8 @@
 @_exported @_documentation(visibility: internal) import struct HTTPTypes.HTTPResponse
 @_exported @_documentation(visibility: internal) import struct HummingbirdCore.BindAddress
 // Temporary exports of unavailable typealiases
-@_exported @_documentation(visibility: internal) import struct HummingbirdCore.HBRequest
-@_exported @_documentation(visibility: internal) import struct HummingbirdCore.HBRequestBody
-@_exported @_documentation(visibility: internal) import struct HummingbirdCore.HBResponse
-@_exported @_documentation(visibility: internal) import struct HummingbirdCore.HBResponseBody
-@_exported @_documentation(visibility: internal) import protocol HummingbirdCore.HBResponseBodyWriter
 @_exported @_documentation(visibility: internal) import struct HummingbirdCore.Request
-@_exported @_documentation(visibility: internal) import struct HummingbirdCore.RequestBody
+@_exported @_documentation(visibility: internal) import class HummingbirdCore.RequestBody
 @_exported @_documentation(visibility: internal) import struct HummingbirdCore.Response
 @_exported @_documentation(visibility: internal) import struct HummingbirdCore.ResponseBody
 @_exported @_documentation(visibility: internal) import protocol HummingbirdCore.ResponseBodyWriter

Sources/Hummingbird/Files/FileIO.swift

@@ -6,6 +6,8 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
+public import AsyncStreaming
+public import BasicContainers
 import CNIOLinux
 public import HummingbirdCore
 import Logging
@@ -105,6 +107,31 @@ public struct FileIO: Sendable {
         }
     }
 
+    /// Write contents of AsyncSequence of buffers to file
+    ///
+    /// - Parameters:
+    ///   - contents: AsyncSequence of buffers to write.
+    ///   - path: Path to write to
+    ///   - context: Request Context
+    public func writeFile<Reader: AsyncReader & ~Copyable>(
+        reader: consuming Reader,
+        path: String,
+        context: some RequestContext
+    ) async throws where Reader.Buffer == UniqueArray<UInt8> {
+        context.logger.debug("[FileIO] PUT", metadata: ["hb.file.path": .string(path)])
+        var reader: Reader? = reader
+        _ = try await self.fileSystem.withFileHandle(
+            forWritingAt: .init(path),
+            options: .newFile(replaceExisting: true)
+        ) { fileHandle in
+            try await fileHandle.withBufferedWriter { writer in
+                try await reader.take()!.forEachBuffer { buffer in
+                    _ = try await writer.write(contentsOf: ByteBuffer(buffer.span.bytes))
+                }
+            }
+        }
+    }
+
     /// Write contents of buffer to file
     ///
     /// - Parameters:

Sources/Hummingbird/Middleware/MetricsMiddleware.swift

@@ -33,9 +33,10 @@ public struct MetricsMiddleware<Context: RequestContext>: RouterMiddleware {
         do {
             var response = try await next(request, context)
             let responseStatus = response.status
+            let method = request.method
             response.body = response.body.withPostWriteClosure {
                 let metrics = self.metricsCache.getEndpointMetrics(
-                    id: .init(endpoint: context.endpointPath ?? "Unknown", method: request.method, status: responseStatus)
+                    id: .init(endpoint: context.endpointPath ?? "Unknown", method: method, status: responseStatus)
                 )
                 metrics.counter.increment()
                 metrics.timer.recordNanoseconds(DispatchTime.now().uptimeNanoseconds - startTime)

Sources/Hummingbird/Server/Request.swift

@@ -16,19 +16,6 @@ public import Foundation
 #endif
 
 extension Request {
-    /// Collapse body into one ByteBuffer.
-    ///
-    /// This will store the collated ByteBuffer back into the request so is a mutating method. If
-    /// you don't need to store the collated ByteBuffer on the request then use
-    /// `request.body.collate(maxSize:)`.
-    ///
-    /// - Parameter context: Request context
-    /// - Returns: Collated body
-    @_documentation(visibility: internal) @available(*, unavailable, message: "Use Request.collectBody(upTo:) instead")
-    public mutating func collateBody(context: some RequestContext) async throws -> ByteBuffer {
-        try await self.collectBody(upTo: context.maxUploadSize)
-    }
-
     /// Decode request using decoder stored at ``Hummingbird/RequestContext/requestDecoder``.
     /// - Parameters
     ///   - type: Type you want to decode to

Sources/HummingbirdCore/Request/Request.swift

@@ -6,41 +6,27 @@
 // SPDX-License-Identifier: Apache-2.0
 //
 
+public import BasicContainers
 public import HTTPTypes
-public import NIOCore
-package import NIOHTTPTypes
+import NIOCore
 
 /// Holds all the values required to process a request
-public struct Request: Sendable {
+public struct Request {
     // MARK: Member variables
 
     /// URI path
     public let uri: URI
     /// HTTP head
     public let head: HTTPRequest
     /// Body of HTTP request
-    private var _body: RequestBody
+    public var body: RequestBody
     /// Request HTTP method
     @inlinable
     public var method: HTTPRequest.Method { self.head.method }
     /// Request HTTP headers
     @inlinable
     public var headers: HTTPFields { self.head.headerFields }
 
-    public var body: RequestBody {
-        get { _body }
-        set {
-            let original = _body.originalRequestBody
-            switch newValue._backing {
-            case .nioAsyncChannelRequestBody:
-                self._body = body
-            case .byteBuffer(let buffer, _):
-                self._body = .init(.byteBuffer(buffer, original))
-            case .anyAsyncSequence(let seq, _):
-                self._body = .init(.anyAsyncSequence(seq, original))
-            }
-        }
-    }
     // MARK: Initialization
 
     /// Create new Request
@@ -53,37 +39,25 @@ public struct Request: Sendable {
     ) {
         self.uri = .init(head.path ?? "")
         self.head = head
-        self._body = body
-    }
-
-    /// Create new Request
-    /// - Parameters:
-    ///   - head: HTTP head
-    ///   - bodyIterator: HTTP request part stream
-    package init(
-        head: HTTPRequest,
-        bodyIterator: NIOAsyncChannelInboundStream<HTTPRequestPart>.AsyncIterator
-    ) {
-        self.uri = .init(head.path ?? "")
-        self.head = head
-        self._body = .init(nioAsyncChannelInbound: .init(iterator: bodyIterator))
+        self.body = body
     }
 
-    /// Collapse body into one ByteBuffer.
+    /// Collapse body into one UniqueArray.
     ///
     /// This will store the collated ByteBuffer back into the request so is a mutating method. If
     /// you don't need to store the collated ByteBuffer on the request then use
     /// `request.body.collect(maxSize:)`.
     ///
-    /// - Parameter maxSize: Maxiumum size of body to collect
-    /// - Returns: Collated body
-    public mutating func collectBody(upTo maxSize: Int) async throws -> ByteBuffer {
-        let byteBuffer = try await self.body.collect(upTo: maxSize)
-        self.body = .init(buffer: byteBuffer)
-        return byteBuffer
+    /// - Parameters
+    ///     - maxSize: Maxiumum size of body to collect
+    public mutating func collectBody(upTo maxSize: Int, process: (inout UniqueArray<UInt8>) async throws -> Void) async throws {
+        var array = try await self.body.collect(upTo: maxSize)
+        try await process(&array)
+        self.body = .init(.asyncReader(CollatedRequestAsyncReader(array)))
     }
 }
 
+@available(hummingbird 3.0, *)
 extension Request: CustomStringConvertible {
     public var description: String {
         "uri: \(self.uri), method: \(self.method), headers: \(self.headers), body: \(self.body)"