Add support for configuring backpressure (#14)

Author: gjcairo

Sources/HTTPServer/HTTPServer.swift

@@ -155,11 +155,21 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
         configuration: HTTPServerConfiguration,
         handler: RequestHandler
     ) async throws {
+        let asyncChannelConfiguration: NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>.Configuration
+        switch configuration.backpressureStrategy.backing {
+        case .watermark(let low, let high):
+            asyncChannelConfiguration = .init(
+                backPressureStrategy: .init(lowWatermark: low, highWatermark: high),
+                isOutboundHalfClosureEnabled: true
+            )
+        }
+
         switch configuration.tlSConfiguration.backing {
         case .insecure:
             try await Self.serveInsecureHTTP1_1(
                 bindTarget: configuration.bindTarget,
                 handler: handler,
+                asyncChannelConfiguration: asyncChannelConfiguration,
                 logger: logger
             )
 
@@ -169,6 +179,7 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
                 certificateChain: certificateChain,
                 privateKey: privateKey,
                 handler: handler,
+                asyncChannelConfiguration: asyncChannelConfiguration,
                 logger: logger
             )
         }
@@ -177,6 +188,7 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
     private static func serveInsecureHTTP1_1(
         bindTarget: HTTPServerConfiguration.BindTarget,
         handler: RequestHandler,
+        asyncChannelConfiguration: NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>.Configuration,
         logger: Logger
     ) async throws {
         switch bindTarget.backing {
@@ -188,7 +200,7 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
                         try channel.pipeline.syncOperations.addHandler(HTTP1ToHTTPServerCodec(secure: false))
                         return try NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>(
                             wrappingChannelSynchronously: channel,
-                            configuration: .init(isOutboundHalfClosureEnabled: true)
+                            configuration: asyncChannelConfiguration
                         )
                     }
                 }
@@ -214,6 +226,7 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
         certificateChain: [Certificate],
         privateKey: Certificate.PrivateKey,
         handler: RequestHandler,
+        asyncChannelConfiguration: NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>.Configuration,
         logger: Logger
     ) async throws {
         switch bindTarget.backing {
@@ -256,7 +269,7 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
 
                                 return try NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>(
                                     wrappingChannelSynchronously: channel,
-                                    configuration: .init(isOutboundHalfClosureEnabled: true)
+                                    configuration: asyncChannelConfiguration
                                 )
                             }
                         } http2ConnectionInitializer: { channel in
@@ -270,7 +283,7 @@ public final class Server<RequestHandler: HTTPServerRequestHandler> {
 
                                 return try NIOAsyncChannel<HTTPRequestPart, HTTPResponsePart>(
                                     wrappingChannelSynchronously: channel,
-                                    configuration: .init(isOutboundHalfClosureEnabled: true)
+                                    configuration: asyncChannelConfiguration
                                 )
                             }
                         }

Sources/HTTPServer/HTTPServerConfiguration.swift

@@ -48,9 +48,7 @@ public struct HTTPServerConfiguration: Sendable {
 
         let backing: Backing
 
-        public static func insecure() -> Self {
-            Self(backing: .insecure)
-        }
+        public static let insecure: Self = Self(backing: .insecure)
 
         public static func certificateChainAndPrivateKey(
             certificateChain: [Certificate],
@@ -65,16 +63,50 @@ public struct HTTPServerConfiguration: Sendable {
         }
     }
 
+    /// Configuration for the backpressure strategy to use when reading requests and writing back responses.
+    public struct BackPressureStrategy: Sendable {
+        enum Backing {
+            case watermark(low: Int, high: Int)
+        }
+
+        internal let backing: Backing
+
+        private init(backing: Backing) {
+            self.backing = backing
+        }
+
+        /// A low/high watermark will be applied when reading requests and writing responses.
+        /// - Parameters:
+        ///   - low: The threshold below which the consumer will ask the producer to produce more elements.
+        ///   - high: The threshold above which the producer will stop producing elements.
+        /// - Returns: A low/high watermark strategy with the configured thresholds.
+        public static func watermark(low: Int, high: Int) -> Self {
+            .init(backing: .watermark(low: low, high: high))
+        }
+    }
+
     /// Network binding configuration
     public var bindTarget: BindTarget
 
+    /// TLS configuration for the server.
     public var tlSConfiguration: TLSConfiguration
 
+    /// Backpressure strategy to use in the server.
+    public var backpressureStrategy: BackPressureStrategy
+
+    /// Create a new configuration.
+    /// - Parameters:
+    ///   - bindTarget: A ``BindTarget``.
+    ///   - tlsConfiguration: A ``TLSConfiguration``. Defaults to ``TLSConfiguration/insecure``.
+    ///   - backpressureStrategy: A ``BackPressureStrategy``.
+    ///   Defaults to ``BackPressureStrategy/watermark(low:high:)`` with a low watermark of 2 and a high of 10.
     public init(
         bindTarget: BindTarget,
-        tlsConfiguration: TLSConfiguration = .insecure()
+        tlsConfiguration: TLSConfiguration = .insecure,
+        backpressureStrategy: BackPressureStrategy = .watermark(low: 2, high: 10)
     ) {
         self.bindTarget = bindTarget
         self.tlSConfiguration = tlsConfiguration
+        self.backpressureStrategy = backpressureStrategy
     }
 }