URLSessionTask: implement InputStream

Foundation/URLSession/BodySource.swift

@@ -57,6 +57,39 @@ internal enum _BodySourceDataChunk {
     case error
 }
 
+internal final class _BodyStreamSource {
+    let inputStream: InputStream
+
+    init(inputStream: InputStream) {
+        self.inputStream = inputStream
+    }
+}
+
+extension _BodyStreamSource : _BodySource {
+    func getNextChunk(withLength length: Int) -> _BodySourceDataChunk {
+        if inputStream.hasBytesAvailable {
+            let buffer = UnsafeMutableRawBufferPointer.allocate(count: length)
+            guard let pointer = buffer.baseAddress?.assumingMemoryBound(to: UInt8.self) else {
+                return .error
+            }
+            let readBytes = self.inputStream.read(pointer, maxLength: length)
+            if readBytes > 0 {
+                let dispatchData = DispatchData(bytes: UnsafeRawBufferPointer(buffer))
+                return .data(dispatchData.subdata(in: 0 ..< readBytes))
+            }
+            else if readBytes == 0 {
+                return .done
+            }
+            else {
+                return .error
+            }
+        }
+        else {
+            return .done
+        }
+    }
+}
+
 /// A body data source backed by `DispatchData`.
 internal final class _BodyDataSource {
     var data: DispatchData! 

Foundation/URLSession/NativeProtocol.swift

@@ -260,8 +260,29 @@ internal class _NativeProtocol: URLProtocol, _EasyHandleDelegate {
     }
 
     func seekInputStream(to position: UInt64) throws {
-        // We will reset the body source and seek forward.
-        NSUnimplemented()
+        // We will reset the body sourse and seek forward.
+        guard let session = task?.session as? URLSession else { fatalError() }
+        if let delegate = session.delegate as? URLSessionTaskDelegate {
+            delegate.urlSession(session, task: task!, needNewBodyStream: { [weak self] inputStream in
+                if let strongSelf = self, let url = strongSelf.request.url, let inputStream = inputStream {
+                    switch strongSelf.internalState {
+                    case .transferInProgress(let currentTransferState):
+                        switch currentTransferState.requestBodySource {
+                        case is _BodyStreamSource:
+                            let drain = strongSelf.createTransferBodyDataDrain()
+                            let source = _BodyStreamSource(inputStream: inputStream)
+                            let transferState = _TransferState(url: url, bodyDataDrain: drain, bodySource: source)
+                            strongSelf.internalState = .transferInProgress(transferState)
+                        default:
+                            NSUnimplemented()
+                        }
+                    default:
+                        //TODO: it's possible?
+                        break
+                    }
+                }
+            })
+        }
     }
 
     func updateProgressMeter(with propgress: _EasyHandle._Progress) {
@@ -313,8 +334,9 @@ internal class _NativeProtocol: URLProtocol, _EasyHandleDelegate {
                 self?.easyHandle.unpauseSend()
             })
             return _TransferState(url: url, bodyDataDrain: drain,bodySource: source)
-        case .stream:
-            NSUnimplemented()
+        case .stream(let inputStream):
+            let source = _BodyStreamSource(inputStream: inputStream)
+            return _TransferState(url: url, bodyDataDrain: drain, bodySource: source)
         }
     }
 

Foundation/URLSession/URLSessionTask.swift

@@ -57,6 +57,8 @@ open class URLSessionTask : NSObject, NSCopying {
     internal convenience init(session: URLSession, request: URLRequest, taskIdentifier: Int) {
         if let bodyData = request.httpBody {
             self.init(session: session, request: request, taskIdentifier: taskIdentifier, body: _Body.data(createDispatchData(bodyData)))
+        }else if let bodyStream = request.httpBodyStream {
+            self.init(session: session, request: request, taskIdentifier: taskIdentifier, body: _Body.stream(bodyStream))
         } else {
             self.init(session: session, request: request, taskIdentifier: taskIdentifier, body: .none)
         }

TestFoundation/TestURLSession.swift

@@ -15,6 +15,7 @@ class TestURLSession : LoopbackServerTest {
             ("test_dataTaskWithURLRequest", test_dataTaskWithURLRequest),
             ("test_dataTaskWithURLCompletionHandler", test_dataTaskWithURLCompletionHandler),
             ("test_dataTaskWithURLRequestCompletionHandler", test_dataTaskWithURLRequestCompletionHandler),
+            ("test_dataTaskWithHttpInputStream", test_dataTaskWithHttpInputStream),
             ("test_downloadTaskWithURL", test_downloadTaskWithURL),
             ("test_downloadTaskWithURLRequest", test_downloadTaskWithURLRequest),
             ("test_downloadTaskWithRequestAndHandler", test_downloadTaskWithRequestAndHandler),
@@ -119,6 +120,56 @@ class TestURLSession : LoopbackServerTest {
         waitForExpectations(timeout: 12)
     }
 
+    func test_dataTaskWithHttpInputStream() {
+        func randomString(length: Int) -> String {
+            let letters = Array("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
+            let len = letters.count
+
+            var randomString = ""
+
+            for _ in 0 ..< length {
+                let rand = Int.random(in: 0..<len)
+                let nextChar = letters[rand]
+                randomString += String(nextChar)
+            }
+            return randomString
+        }
+
+        let delegate = HTTPBinResponseDelegateJSON<HTTPBinResponse>()
+
+        let dataString = randomString(length: 65537)
+
+        let urlString = "http://httpbin.org/post"
+        let url = URL(string: urlString)!
+        let urlSession = URLSession(configuration: URLSessionConfiguration.default, delegate: delegate, delegateQueue: nil)
+
+        var urlRequest = URLRequest(url: url)
+        urlRequest.httpMethod = "POST"
+
+        guard let data = dataString.data(using: .utf8) else {
+            XCTFail()
+            return
+        }
+
+        let inputStream = InputStream(data: data)
+        inputStream.open()
+
+        urlRequest.httpBodyStream = inputStream
+
+        urlRequest.setValue("en-us", forHTTPHeaderField: "Accept-Language")
+        urlRequest.setValue("text/xml; charset=utf-8", forHTTPHeaderField: "Content-Type")
+        urlRequest.setValue("chunked", forHTTPHeaderField: "Transfer-Encoding")
+
+        let urlTask = urlSession.dataTask(with: urlRequest)
+        urlTask.resume()
+
+        delegate.semaphore.wait()
+
+        XCTAssertTrue(urlTask.response != nil)
+        XCTAssertTrue(delegate.response != nil)
+        XCTAssertTrue(delegate.response?.data == dataString)
+    }
+
     func test_downloadTaskWithURL() {
         let urlString = "http://127.0.0.1:\(TestURLSession.serverPort)/country.txt"
         let url = URL(string: urlString)!
@@ -879,6 +930,51 @@ class HTTPRedirectionDataTask : NSObject {
     }
 }
 
+
+class HTTPBinResponseDelegate<T>: NSObject, URLSessionDataDelegate, URLSessionTaskDelegate {
+    let semaphore = DispatchSemaphore(value: 0)
+    let outputStream = OutputStream.toMemory()
+    var response: T?
+
+    override init() {
+        outputStream.open()
+    }
+
+    deinit {
+        outputStream.close()
+    }
+
+    public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
+        _ = data.withUnsafeBytes({ (bytes: UnsafePointer<UInt8>) in
+            outputStream.write(bytes, maxLength: data.count)
+        })
+    }
+
+    public func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
+        if let data = outputStream.property(forKey: .dataWrittenToMemoryStreamKey) as? NSData {
+            response = parseResposne(data: data._bridgeToSwift())
+        }
+        semaphore.signal()
+    }
+
+    public func parseResposne(data: Data) -> T? {
+        fatalError("")
+    }
+}
+
+class HTTPBinResponseDelegateJSON<T: Codable>: HTTPBinResponseDelegate<T> {
+    override func parseResposne(data: Data) -> T? {
+        return try? JSONDecoder().decode(T.self, from: data)
+    }
+}
+
+struct HTTPBinResponse: Codable {
+    let data: String
+    let headers: [String: String]
+    let origin: String
+    let url: String
+}
+
 extension HTTPRedirectionDataTask : URLSessionDataDelegate {
     public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive response: URLResponse, completionHandler: @escaping (URLSession.ResponseDisposition) -> Void) {
         guard let httpresponse = response as? HTTPURLResponse else { fatalError() }