add response decompression support

Package.swift

@@ -23,11 +23,12 @@ let package = Package(
     dependencies: [
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.8.0"),
         .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.0.0"),
+        .package(url: "https://github.com/apple/swift-nio-extras.git", .branch("master")),
     ],
     targets: [
         .target(
             name: "AsyncHTTPClient",
-            dependencies: ["NIO", "NIOHTTP1", "NIOSSL", "NIOConcurrencyHelpers"]
+            dependencies: ["NIO", "NIOHTTP1", "NIOSSL", "NIOConcurrencyHelpers", "NIOHTTPCompression"]
         ),
         .testTarget(
             name: "AsyncHTTPClientTests",

Sources/AsyncHTTPClient/HTTPClient.swift

@@ -16,6 +16,7 @@ import Foundation
 import NIO
 import NIOConcurrencyHelpers
 import NIOHTTP1
+import NIOHTTPCompression
 import NIOSSL
 
 /// HTTPClient class provides API for request execution.
@@ -252,6 +253,13 @@ public class HTTPClient {
                     case .some(let proxy):
                         return channel.pipeline.addProxyHandler(for: request, decoder: decoder, encoder: encoder, tlsConfiguration: self.configuration.tlsConfiguration, proxy: proxy)
                     }
+                }.flatMap {
+                    switch self.configuration.decompression {
+                    case .disabled:
+                        return channel.eventLoop.makeSucceededFuture(())
+                    case .enabled(let limit):
+                        return channel.pipeline.addHandler(NIOHTTPResponseDecompressor(limit: limit))
+                    }
                 }.flatMap {
                     if let timeout = self.resolve(timeout: self.configuration.timeout.read, deadline: deadline) {
                         return channel.pipeline.addHandler(IdleStateHandler(readTimeout: timeout))
@@ -322,31 +330,37 @@ public class HTTPClient {
         public var timeout: Timeout
         /// Upstream proxy, defaults to no proxy.
         public var proxy: Proxy?
+        /// Enables automatic body decompression. Supported algorithms are gzip and deflate.
+        public var decompression: Decompression
         /// Ignore TLS unclean shutdown error, defaults to `false`.
         public var ignoreUncleanSSLShutdown: Bool
 
-        public init(tlsConfiguration: TLSConfiguration? = nil, followRedirects: Bool = false, timeout: Timeout = Timeout(), proxy: Proxy? = nil) {
-            self.init(tlsConfiguration: tlsConfiguration, followRedirects: followRedirects, timeout: timeout, proxy: proxy, ignoreUncleanSSLShutdown: false)
-        }
-
-        public init(tlsConfiguration: TLSConfiguration? = nil, followRedirects: Bool = false, timeout: Timeout = Timeout(), proxy: Proxy? = nil, ignoreUncleanSSLShutdown: Bool = false) {
+        public init(tlsConfiguration: TLSConfiguration? = nil,
+                    followRedirects: Bool = false,
+                    timeout: Timeout = Timeout(),
+                    proxy: Proxy? = nil,
+                    ignoreUncleanSSLShutdown: Bool = false,
+                    decompression: Decompression = .disabled) {
             self.tlsConfiguration = tlsConfiguration
             self.followRedirects = followRedirects
             self.timeout = timeout
             self.proxy = proxy
             self.ignoreUncleanSSLShutdown = ignoreUncleanSSLShutdown
+            self.decompression = decompression
         }
 
-        public init(certificateVerification: CertificateVerification, followRedirects: Bool = false, timeout: Timeout = Timeout(), proxy: Proxy? = nil) {
-            self.init(certificateVerification: certificateVerification, followRedirects: followRedirects, timeout: timeout, proxy: proxy, ignoreUncleanSSLShutdown: false)
-        }
-
-        public init(certificateVerification: CertificateVerification, followRedirects: Bool = false, timeout: Timeout = Timeout(), proxy: Proxy? = nil, ignoreUncleanSSLShutdown: Bool = false) {
+        public init(certificateVerification: CertificateVerification,
+                    followRedirects: Bool = false,
+                    timeout: Timeout = Timeout(),
+                    proxy: Proxy? = nil,
+                    ignoreUncleanSSLShutdown: Bool = false,
+                    decompression: Decompression = .disabled) {
             self.tlsConfiguration = TLSConfiguration.forClient(certificateVerification: certificateVerification)
             self.followRedirects = followRedirects
             self.timeout = timeout
             self.proxy = proxy
             self.ignoreUncleanSSLShutdown = ignoreUncleanSSLShutdown
+            self.decompression = decompression
         }
     }
 
@@ -403,6 +417,14 @@ public class HTTPClient {
             return EventLoopPreference(.delegateAndChannel(on: eventLoop))
         }
     }
+
+    /// Specifies decompression settings.
+    public enum Decompression {
+        /// Decompression is disabled.
+        case disabled
+        /// Decompression is enabled.
+        case enabled(limit: NIOHTTPDecompression.DecompressionLimit)
+    }
 }
 
 extension HTTPClient.Configuration {
@@ -472,6 +494,9 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
         case invalidProxyResponse
         case contentLengthMissing
         case proxyAuthenticationRequired
+        case decompressionLimit
+        case decompressionInitialization(Int32)
+        case decompression(Int32)
     }
 
     private var code: Code
@@ -508,6 +533,12 @@ public struct HTTPClientError: Error, Equatable, CustomStringConvertible {
     public static let invalidProxyResponse = HTTPClientError(code: .invalidProxyResponse)
     /// Request does not contain `Content-Length` header.
     public static let contentLengthMissing = HTTPClientError(code: .contentLengthMissing)
-    /// Proxy Authentication Required
+    /// Proxy Authentication Required.
     public static let proxyAuthenticationRequired = HTTPClientError(code: .proxyAuthenticationRequired)
+    /// Decompression limit reached.
+    public static let decompressionLimit = HTTPClientError(code: .decompressionLimit)
+    /// Decompression initialization failed.
+    public static func decompressionInitialization(_ code: Int32) -> HTTPClientError { return HTTPClientError(code: .decompressionInitialization(code)) }
+    /// Decompression failed.
+    public static func decompression(_ code: Int32) -> HTTPClientError { return HTTPClientError(code: .decompression(code)) }
 }

Sources/AsyncHTTPClient/HTTPHandler.swift

@@ -677,7 +677,7 @@ extension TaskHandler: ChannelDuplexHandler {
     }
 
     func channelRead(context: ChannelHandlerContext, data: NIOAny) {
-        let response = unwrapInboundIn(data)
+        let response = self.unwrapInboundIn(data)
         switch response {
         case .head(let head):
             if let redirectURL = redirectHandler?.redirectTarget(status: head.status, headers: head.headers) {

Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift

@@ -17,6 +17,7 @@ import Foundation
 import NIO
 import NIOConcurrencyHelpers
 import NIOHTTP1
+import NIOHTTPCompression
 import NIOSSL
 
 class TestHTTPDelegate: HTTPClientResponseDelegate {
@@ -111,30 +112,40 @@ internal final class HTTPBin {
         return channel.pipeline.addHandler(try! NIOSSLServerHandler(context: context), position: .first)
     }
 
-    init(ssl: Bool = false, simulateProxy: HTTPProxySimulator.Option? = nil, channelPromise: EventLoopPromise<Channel>? = nil) {
+    init(ssl: Bool = false, compress: Bool = false, simulateProxy: HTTPProxySimulator.Option? = nil, channelPromise: EventLoopPromise<Channel>? = nil) {
         self.serverChannel = try! ServerBootstrap(group: self.group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .childChannelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
             .childChannelInitializer { channel in
-                channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: true, withErrorHandling: true).flatMap {
-                    if let simulateProxy = simulateProxy {
-                        let responseEncoder = HTTPResponseEncoder()
-                        let requestDecoder = ByteToMessageHandler(HTTPRequestDecoder(leftOverBytesStrategy: .forwardBytes))
-
-                        return channel.pipeline.addHandlers([responseEncoder, requestDecoder, HTTPProxySimulator(option: simulateProxy, encoder: responseEncoder, decoder: requestDecoder)], position: .first)
-                    } else {
-                        return channel.eventLoop.makeSucceededFuture(())
+                channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: true, withErrorHandling: true)
+                    .flatMap {
+                        if compress {
+                            return channel.pipeline.addHandler(HTTPResponseCompressor())
+                        } else {
+                            return channel.eventLoop.makeSucceededFuture(())
+                        }
                     }
-                }.flatMap {
-                    if ssl {
-                        return HTTPBin.configureTLS(channel: channel).flatMap {
-                            channel.pipeline.addHandler(HttpBinHandler(channelPromise: channelPromise))
+                    .flatMap {
+                        if let simulateProxy = simulateProxy {
+                            let responseEncoder = HTTPResponseEncoder()
+                            let requestDecoder = ByteToMessageHandler(HTTPRequestDecoder(leftOverBytesStrategy: .forwardBytes))
+
+                            return channel.pipeline.addHandlers([responseEncoder, requestDecoder, HTTPProxySimulator(option: simulateProxy, encoder: responseEncoder, decoder: requestDecoder)], position: .first)
+                        } else {
+                            return channel.eventLoop.makeSucceededFuture(())
                         }
-                    } else {
-                        return channel.pipeline.addHandler(HttpBinHandler(channelPromise: channelPromise))
                     }
-                }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+                    .flatMap {
+                        if ssl {
+                            return HTTPBin.configureTLS(channel: channel).flatMap {
+                                channel.pipeline.addHandler(HttpBinHandler(channelPromise: channelPromise))
+                            }
+                        } else {
+                            return channel.pipeline.addHandler(HttpBinHandler(channelPromise: channelPromise))
+                        }
+                    }
+            }
+            .bind(host: "127.0.0.1", port: 0).wait()
     }
 
     func shutdown() throws {
@@ -461,6 +472,12 @@ extension ByteBuffer {
         buffer.writeString(string)
         return buffer
     }
+
+    public static func of(bytes: [UInt8]) -> ByteBuffer {
+        var buffer = ByteBufferAllocator().buffer(capacity: bytes.count)
+        buffer.writeBytes(bytes)
+        return buffer
+    }
 }
 
 private let cert = """

Tests/AsyncHTTPClientTests/HTTPClientTests+XCTest.swift

@@ -57,6 +57,8 @@ extension HTTPClientTests {
             ("testWrongContentLengthForSSLUncleanShutdown", testWrongContentLengthForSSLUncleanShutdown),
             ("testWrongContentLengthWithIgnoreErrorForSSLUncleanShutdown", testWrongContentLengthWithIgnoreErrorForSSLUncleanShutdown),
             ("testEventLoopArgument", testEventLoopArgument),
+            ("testDecompression", testDecompression),
+            ("testDecompressionLimit", testDecompressionLimit),
         ]
     }
 }

Tests/AsyncHTTPClientTests/HTTPClientTests.swift

@@ -563,4 +563,60 @@ class HTTPClientTests: XCTestCase {
         response = try httpClient.execute(request: request, delegate: delegate, eventLoop: .delegate(on: eventLoop)).wait()
         XCTAssertEqual(true, response)
     }
+
+    func testDecompression() throws {
+        let httpBin = HTTPBin(compress: true)
+        let httpClient = HTTPClient(eventLoopGroupProvider: .createNew, configuration: .init(decompression: .enabled(limit: .none)))
+        defer {
+            XCTAssertNoThrow(try httpClient.syncShutdown())
+            XCTAssertNoThrow(try httpBin.shutdown())
+        }
+
+        var body = ""
+        for _ in 1...1000 {
+            body += "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua."
+        }
+
+        for algorithm in [nil, "gzip", "deflate"] {
+            var request = try HTTPClient.Request(url: "http://localhost:\(httpBin.port)/post", method: .POST)
+            request.body = .string(body)
+            if let algorithm = algorithm {
+                request.headers.add(name: "Accept-Encoding", value: algorithm)
+            }
+
+            let response = try httpClient.execute(request: request).wait()
+            let bytes = response.body!.getData(at: 0, length: response.body!.readableBytes)!
+            let data = try JSONDecoder().decode(RequestInfo.self, from: bytes)
+
+            XCTAssertEqual(.ok, response.status)
+            XCTAssertGreaterThan(body.count, response.headers["Content-Length"].first.flatMap { Int($0) }!)
+            if let algorithm = algorithm {
+                XCTAssertEqual(algorithm, response.headers["Content-Encoding"].first)
+            } else {
+                XCTAssertEqual("deflate", response.headers["Content-Encoding"].first)
+            }
+            XCTAssertEqual(body, data.data)
+        }
+    }
+
+    func testDecompressionLimit() throws {
+        let httpBin = HTTPBin(compress: true)
+        let httpClient = HTTPClient(eventLoopGroupProvider: .createNew, configuration: .init(decompression: .enabled(limit: .ratio(10))))
+        defer {
+            XCTAssertNoThrow(try httpClient.syncShutdown())
+            XCTAssertNoThrow(try httpBin.shutdown())
+        }
+
+        var request = try HTTPClient.Request(url: "http://localhost:\(httpBin.port)/post", method: .POST)
+        request.body = .byteBuffer(ByteBuffer.of(bytes: [120, 156, 75, 76, 28, 5, 200, 0, 0, 248, 66, 103, 17]))
+        request.headers.add(name: "Accept-Encoding", value: "deflate")
+
+        do {
+            _ = try httpClient.execute(request: request).wait()
+        } catch let error as HTTPClientError {
+            XCTAssertEqual(error, .decompressionLimit)
+        } catch {
+            XCTFail("Unexptected error: \(error)")
+        }
+    }
 }

docker/Dockerfile

@@ -13,7 +13,7 @@ ENV LANGUAGE en_US.UTF-8
 
 # dependencies
 RUN apt-get update && apt-get install -y wget
-RUN apt-get update && apt-get install -y lsof dnsutils netcat-openbsd net-tools # used by integration tests
+RUN apt-get update && apt-get install -y lsof dnsutils netcat-openbsd net-tools libz-dev # used by integration tests
 
 # ruby and jazzy for docs generation
 RUN apt-get update && apt-get install -y ruby ruby-dev libsqlite3-dev