Implement seek InputStream

Author: albertaleksieiev

Foundation/Stream.swift

@@ -114,7 +114,10 @@ open class Stream: NSObject {
 // InputStream is an abstract class representing the base functionality of a read stream.
 // Subclassers are required to implement these methods.
 open class InputStream: Stream {
-
+    enum _Error: Error {
+        case cantSeekInputStream
+    }
+    
     internal let _stream: CFReadStream!
 
     // reads up to length bytes into the supplied buffer, which must be at least of size len. Returns the actual number of bytes read.
@@ -225,6 +228,43 @@ open class OutputStream : Stream {
     }
 }
 
+extension InputStream {
+    func seek(to position: UInt64) throws {
+        guard position > 0 else {
+            return
+        }
+        
+        guard position < Int.max else { throw _Error.cantSeekInputStream }
+        
+        let bufferSize = 1024
+        var remainingBytes = Int(position)
+        
+        let buffer = UnsafeMutableRawBufferPointer.allocate(byteCount: bufferSize, alignment: MemoryLayout<UInt8>.alignment)
+        
+        guard let pointer = buffer.baseAddress?.assumingMemoryBound(to: UInt8.self) else {
+            buffer.deallocate()
+            throw _Error.cantSeekInputStream
+        }
+        
+        if self.streamStatus == .notOpen {
+            self.open()
+        }
+        
+        while remainingBytes > 0 && self.hasBytesAvailable {
+            let read = self.read(pointer, maxLength: min(bufferSize, remainingBytes))
+            if read == -1 {
+                throw _Error.cantSeekInputStream
+            }
+            remainingBytes -= read
+        }
+        
+        buffer.deallocate()
+        if remainingBytes != 0 {
+            throw _Error.cantSeekInputStream
+        }
+    }
+}
+
 // Discussion of this API is ongoing for its usage of AutoreleasingUnsafeMutablePointer
 #if false
 extension Stream {

Foundation/URLSession/NativeProtocol.swift

@@ -269,6 +269,7 @@ internal class _NativeProtocol: URLProtocol, _EasyHandleDelegate {
                     case .transferInProgress(let currentTransferState):
                         switch currentTransferState.requestBodySource {
                         case is _BodyStreamSource:
+                            try inputStream.seek(to: position)
                             let drain = strongSelf.createTransferBodyDataDrain()
                             let source = _BodyStreamSource(inputStream: inputStream)
                             let transferState = _TransferState(url: url, bodyDataDrain: drain, bodySource: source)

TestFoundation/TestStream.swift

@@ -7,6 +7,23 @@
 // See http://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
 //
 
+private extension Data {
+    init(reading input: InputStream) {
+        self.init()
+        input.open()
+        
+        let bufferSize = 1024
+        let buffer = UnsafeMutablePointer<UInt8>.allocate(capacity: bufferSize)
+        while input.hasBytesAvailable {
+            let read = input.read(buffer, maxLength: bufferSize)
+            self.append(buffer, count: read)
+        }
+        buffer.deallocate(capacity: bufferSize)
+        
+        input.close()
+    }
+}
+
 class TestStream : XCTestCase {
     static var allTests: [(String, (TestStream) -> () throws -> Void)] {
         return [
@@ -15,6 +32,7 @@ class TestStream : XCTestCase {
             ("test_InputStreamWithFile", test_InputStreamWithFile),
             ("test_InputStreamHasBytesAvailable", test_InputStreamHasBytesAvailable),
             ("test_InputStreamInvalidPath", test_InputStreamInvalidPath),
+            ("test_InputStreamSeekToPosition", test_InputStreamSeekToPosition),
             ("test_outputStreamCreationToFile", test_outputStreamCreationToFile),
             ("test_outputStreamCreationToBuffer", test_outputStreamCreationToBuffer),
             ("test_outputStreamCreationWithUrl", test_outputStreamCreationWithUrl),
@@ -116,6 +134,50 @@ class TestStream : XCTestCase {
         XCTAssertEqual(.error, fileStream.streamStatus)
     }
 
+    func test_InputStreamSeekToPosition() {
+        let str = "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Cras congue laoreet facilisis. Sed porta tristique orci. Fusce ut nisl dignissim, tempor tortor id, molestie neque. Nam non tincidunt mi. Integer ac diam quis leo aliquam congue et non magna. In porta mauris suscipit erat pulvinar, sed fringilla quam ornare. Nulla vulputate et ligula vitae sollicitudin. Nulla vel vehicula risus. Quisque eu urna ullamcorper, tincidunt ante vitae, aliquet sem. Suspendisse nec turpis placerat, porttitor ex vel, tristique orci. Maecenas pretium, augue non elementum imperdiet, diam ex vestibulum tortor, non ultrices ante enim iaculis ex. Fusce ut nisl dignissim, tempor tortor id, molestie neque. Nam non tincidunt mi. Integer ac diam quis leo aliquam congue et non magna. In porta mauris suscipit erat pulvinar, sed fringilla quam ornare. Nulla vulputate et ligula vitae sollicitudin. Nulla vel vehicula risus. Quisque eu urna ullamcorper, tincidunt ante vitae, aliquet sem. Suspendisse nec turpis placerat, porttitor ex vel, tristique orci. Maecenas pretium, augue non elementum imperdiet, diam ex vestibulum tortor, non ultrices ante enim iaculis ex.Lorem ipsum dolor sit amet, consectetur adipiscing elit. Cras congue laoreet facilisis. Sed porta tristique orci. Fusce ut nisl dignissim, tempor tortor id, molestie neque. Nam non tincidunt mi. Integer ac diam quis leo aliquam congue et non magna. In porta mauris suscipit erat pulvinar, sed fringilla quam ornare. Nulla vulputate et ligula vitae sollicitudin. Nulla vel vehicula risus. Quisque eu urna ullamcorper, tincidunt ante vitae, aliquet sem. Suspendisse nec turpis placerat, porttitor ex vel."
+        XCTAssert(str.count > 1024) // str.count must be bigger than buffersize inside InputStream.seek func.
+        
+        func testSubdata(_ pos: UInt64) throws -> Data? {
+            guard let data = str.data(using: .utf8) else {
+                XCTFail()
+                return nil
+            }
+            
+            let stream = InputStream(data: data)
+            stream.open()
+            
+            try stream.seek(to: pos)
+            let streamData = Data(reading: stream)
+            
+            let subdata = data.subdata(in: Range(Int(pos)..<data.count))
+            XCTAssertEqual(streamData, subdata)
+            
+            return subdata
+        }
+        
+        var sum = 0
+        for i in 0...str.count {
+            do {
+                sum += try testSubdata(UInt64(i))!.count
+            } catch _ {
+                XCTFail()
+            }
+        }
+        
+        XCTAssertEqual(((1 + str.count) * str.count)/2, sum) // Test on sum of arithmetic sequence :)
+        XCTAssertEqual(try testSubdata(UInt64(str.count))!.count, 0) // It shouldbe end
+        
+        do {
+            try testSubdata(UInt64(str.count + 1)) // out of boundaries
+            XCTFail()
+        } catch let error as InputStream._Error {
+            XCTAssertEqual(error, .cantSeekInputStream)
+        } catch {
+            XCTFail()
+        }
+    }
+    
     func test_outputStreamCreationToFile() {
         guard let filePath = createTestFile("TestFileOut.txt", _contents: Data(capacity: 256)) else {
             XCTFail("Unable to create temp file");