Add FileDownloadDelegate for simple file downloads (swift-server#275)

* Add FileDownloadDelegate for simple file downloads

* Add testFileDownload to the allTests array

* Fix formatting

* Fix compatibility with Swift 5.0

* Add doc comments, update README.md

* Refine FileDownloadDelegate description in README

* Bump NIO version, remove weak self, cleanup test

* Fix formatting issues in a doc comment

* Create separate Progress struct, async open file

* Create an ad-hoc EventLoopGroup for opening a file

* Move file opening code to `didReceiveBodyPart`

* Fix linter error in FileDownloadDelegate.swift

* Fix wrong future assignment in FileDownloadDelegate

* Fix Swift 5.0 return statement compatibility

* Fix linter warning

* Fix Swift 5.0 return statement compatibility

* Remove redundant `write` function

* Add negative test case and separate testing endpoint

* Add missing testFileDownloadError to the manifest

Author: MaxDesiatov

README.md

@@ -165,6 +165,35 @@ httpClient.execute(request: request, delegate: delegate).futureResult.whenSucces
 }
 ```
 
+### File downloads
+
+Based on the `HTTPClientResponseDelegate` example above you can build more complex delegates,
+the built-in `FileDownloadDelegate` is one of them. It allows streaming the downloaded data
+asynchronously, while reporting the download progress at the same time, like in the following
+example:
+
+```swift
+let client = HTTPClient(eventLoopGroupProvider: .createNew)
+let request = try HTTPClient.Request(
+    url: "https://swift.org/builds/development/ubuntu1804/latest-build.yml"
+)
+
+let delegate = try FileDownloadDelegate(path: "/tmp/latest-build.yml", reportProgress: {
+    if let totalSize = $0 {
+        print("Total bytes count: \(totalSize)")
+    }
+    print("Downloaded \($1) bytes so far")
+})
+
+client.execute(request: request, delegate: delegate).futureResult
+    .whenSuccess { finalTotalBytes, downloadedBytes in
+        if let totalSize = $0 {
+            print("Final total bytes count: \(totalSize)")
+        }
+        print("Downloaded finished with \($1) bytes downloaded")
+    }
+```
+
 ### Unix Domain Socket Paths
 Connecting to servers bound to socket paths is easy:
 ```swift

Sources/AsyncHTTPClient/FileDownloadDelegate.swift

@@ -0,0 +1,129 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the AsyncHTTPClient open source project
+//
+// Copyright (c) 2020 Apple Inc. and the AsyncHTTPClient project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of AsyncHTTPClient project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIO
+import NIOHTTP1
+
+/// Handles a streaming download to a given file path, allowing headers and progress to be reported.
+public final class FileDownloadDelegate: HTTPClientResponseDelegate {
+    /// The response type for this delegate: the total count of bytes as reported by the response
+    /// "Content-Length" header (if available) and the count of bytes downloaded.
+    public struct Progress {
+        public var totalBytes: Int?
+        public var receivedBytes: Int
+    }
+
+    private var progress = Progress(totalBytes: nil, receivedBytes: 0)
+
+    public typealias Response = Progress
+
+    private let filePath: String
+    private let io: NonBlockingFileIO
+    private let reportHead: ((HTTPResponseHead) -> Void)?
+    private let reportProgress: ((Progress) -> Void)?
+
+    private var fileHandleFuture: EventLoopFuture<NIOFileHandle>?
+    private var writeFuture: EventLoopFuture<Void>?
+
+    /// Initializes a new file download delegate.
+    /// - parameters:
+    ///     - path: Path to a file you'd like to write the download to.
+    ///     - pool: A thread pool to use for asynchronous file I/O.
+    ///     - reportHead: A closure called when the response head is available.
+    ///     - reportProgress: A closure called when a body chunk has been downloaded, with
+    ///       the total byte count and download byte count passed to it as arguments. The callbacks
+    ///       will be invoked in the same threading context that the delegate itself is invoked,
+    ///       as controlled by `EventLoopPreference`.
+    public init(
+        path: String,
+        pool: NIOThreadPool = NIOThreadPool(numberOfThreads: 1),
+        reportHead: ((HTTPResponseHead) -> Void)? = nil,
+        reportProgress: ((Progress) -> Void)? = nil
+    ) throws {
+        pool.start()
+        self.io = NonBlockingFileIO(threadPool: pool)
+        self.filePath = path
+
+        self.reportHead = reportHead
+        self.reportProgress = reportProgress
+    }
+
+    public func didReceiveHead(
+        task: HTTPClient.Task<Response>,
+        _ head: HTTPResponseHead
+    ) -> EventLoopFuture<Void> {
+        self.reportHead?(head)
+
+        if let totalBytesString = head.headers.first(name: "Content-Length"),
+            let totalBytes = Int(totalBytesString) {
+            self.progress.totalBytes = totalBytes
+        }
+
+        return task.eventLoop.makeSucceededFuture(())
+    }
+
+    public func didReceiveBodyPart(
+        task: HTTPClient.Task<Response>,
+        _ buffer: ByteBuffer
+    ) -> EventLoopFuture<Void> {
+        self.progress.receivedBytes += buffer.readableBytes
+        self.reportProgress?(self.progress)
+
+        let writeFuture: EventLoopFuture<Void>
+        if let fileHandleFuture = self.fileHandleFuture {
+            writeFuture = fileHandleFuture.flatMap {
+                self.io.write(fileHandle: $0, buffer: buffer, eventLoop: task.eventLoop)
+            }
+        } else {
+            let fileHandleFuture = self.io.openFile(
+                path: self.filePath,
+                mode: .write,
+                flags: .allowFileCreation(),
+                eventLoop: task.eventLoop
+            )
+            self.fileHandleFuture = fileHandleFuture
+            writeFuture = fileHandleFuture.flatMap {
+                self.io.write(fileHandle: $0, buffer: buffer, eventLoop: task.eventLoop)
+            }
+        }
+
+        self.writeFuture = writeFuture
+        return writeFuture
+    }
+
+    private func close(fileHandle: NIOFileHandle) {
+        try! fileHandle.close()
+        self.fileHandleFuture = nil
+    }
+
+    private func finalize() {
+        if let writeFuture = self.writeFuture {
+            writeFuture.whenComplete { _ in
+                self.fileHandleFuture?.whenSuccess(self.close(fileHandle:))
+                self.writeFuture = nil
+            }
+        } else {
+            self.fileHandleFuture?.whenSuccess(self.close(fileHandle:))
+        }
+    }
+
+    public func didReceiveError(task: HTTPClient.Task<Progress>, _ error: Error) {
+        self.finalize()
+    }
+
+    public func didFinishRequest(task: HTTPClient.Task<Response>) throws -> Response {
+        self.finalize()
+        return self.progress
+    }
+}

Tests/AsyncHTTPClientTests/HTTPClientTestUtils.swift

@@ -182,6 +182,31 @@ enum TemporaryFileHelpers {
         }
         return try body(shortEnoughPath)
     }
+
+    /// This function creates a filename that can be used as a temporary file.
+    internal static func withTemporaryFilePath<T>(
+        directory: String = temporaryDirectory,
+        _ body: (String) throws -> T
+    ) throws -> T {
+        let (fd, path) = self.openTemporaryFile()
+        close(fd)
+        try! FileManager.default.removeItem(atPath: path)
+
+        defer {
+            if FileManager.default.fileExists(atPath: path) {
+                try? FileManager.default.removeItem(atPath: path)
+            }
+        }
+        return try body(path)
+    }
+
+    internal static func fileSize(path: String) throws -> Int? {
+        return try FileManager.default.attributesOfItem(atPath: path)[.size] as? Int
+    }
+
+    internal static func fileExists(path: String) -> Bool {
+        return FileManager.default.fileExists(atPath: path)
+    }
 }
 
 internal final class HTTPBin {
@@ -420,6 +445,25 @@ internal final class HttpBinHandler: ChannelInboundHandler {
         }
     }
 
+    func writeEvents(context: ChannelHandlerContext, isContentLengthRequired: Bool = false) {
+        let headers: HTTPHeaders
+        if isContentLengthRequired {
+            headers = HTTPHeaders([("Content-Length", "50")])
+        } else {
+            headers = HTTPHeaders()
+        }
+
+        context.write(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .ok, headers: headers))), promise: nil)
+        for i in 0..<10 {
+            let msg = "id: \(i)"
+            var buf = context.channel.allocator.buffer(capacity: msg.count)
+            buf.writeString(msg)
+            context.writeAndFlush(wrapOutboundOut(.body(.byteBuffer(buf))), promise: nil)
+            Thread.sleep(forTimeInterval: 0.05)
+        }
+        context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
+    }
+
     func channelRead(context: ChannelHandlerContext, data: NIOAny) {
         self.isServingRequest = true
         switch self.unwrapInboundIn(data) {
@@ -531,16 +575,10 @@ internal final class HttpBinHandler: ChannelInboundHandler {
                 context.writeAndFlush(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .ok))), promise: nil)
                 return
             case "/events/10/1": // TODO: parse path
-                context.write(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .ok))), promise: nil)
-                for i in 0..<10 {
-                    let msg = "id: \(i)"
-                    var buf = context.channel.allocator.buffer(capacity: msg.count)
-                    buf.writeString(msg)
-                    context.writeAndFlush(wrapOutboundOut(.body(.byteBuffer(buf))), promise: nil)
-                    Thread.sleep(forTimeInterval: 0.05)
-                }
-                context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
+                self.writeEvents(context: context)
                 return
+            case "/events/10/content-length":
+                self.writeEvents(context: context, isContentLengthRequired: true)
             default:
                 context.write(wrapOutboundOut(.head(HTTPResponseHead(version: HTTPVersion(major: 1, minor: 1), status: .notFound))), promise: nil)
                 context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)

Tests/AsyncHTTPClientTests/HTTPClientTests+XCTest.swift

@@ -44,6 +44,8 @@ extension HTTPClientTests {
             ("testPercentEncodedBackslash", testPercentEncodedBackslash),
             ("testMultipleContentLengthHeaders", testMultipleContentLengthHeaders),
             ("testStreaming", testStreaming),
+            ("testFileDownload", testFileDownload),
+            ("testFileDownloadError", testFileDownloadError),
             ("testRemoteClose", testRemoteClose),
             ("testReadTimeout", testReadTimeout),
             ("testConnectTimeout", testConnectTimeout),

Tests/AsyncHTTPClientTests/HTTPClientTests.swift

@@ -488,6 +488,54 @@ class HTTPClientTests: XCTestCase {
         XCTAssertEqual(10, count)
     }
 
+    func testFileDownload() throws {
+        var request = try Request(url: self.defaultHTTPBinURLPrefix + "events/10/content-length")
+        request.headers.add(name: "Accept", value: "text/event-stream")
+
+        let progress =
+            try TemporaryFileHelpers.withTemporaryFilePath { path -> FileDownloadDelegate.Progress in
+                let delegate = try FileDownloadDelegate(path: path)
+
+                let progress = try self.defaultClient.execute(
+                    request: request,
+                    delegate: delegate
+                )
+                .wait()
+
+                try XCTAssertEqual(50, TemporaryFileHelpers.fileSize(path: path))
+
+                return progress
+            }
+
+        XCTAssertEqual(50, progress.totalBytes)
+        XCTAssertEqual(50, progress.receivedBytes)
+    }
+
+    func testFileDownloadError() throws {
+        var request = try Request(url: self.defaultHTTPBinURLPrefix + "not-found")
+        request.headers.add(name: "Accept", value: "text/event-stream")
+
+        let progress =
+            try TemporaryFileHelpers.withTemporaryFilePath { path -> FileDownloadDelegate.Progress in
+                let delegate = try FileDownloadDelegate(path: path, reportHead: {
+                    XCTAssertEqual($0.status, .notFound)
+                })
+
+                let progress = try self.defaultClient.execute(
+                    request: request,
+                    delegate: delegate
+                )
+                .wait()
+
+                XCTAssertFalse(TemporaryFileHelpers.fileExists(path: path))
+
+                return progress
+            }
+
+        XCTAssertEqual(nil, progress.totalBytes)
+        XCTAssertEqual(0, progress.receivedBytes)
+    }
+
     func testRemoteClose() throws {
         XCTAssertThrowsError(try self.defaultClient.get(url: self.defaultHTTPBinURLPrefix + "close").wait(), "Should fail") { error in
             guard case let error = error as? HTTPClientError, error == .remoteConnectionClosed else {