add pipe to stream

Author: paulofaria

Aeon/HTTPServer/TCPStream.swift

@@ -28,8 +28,8 @@ import Stream
 
 final class TCPStream: StreamType {
     let socket: TCPClientSocket
-    let channel: IOChannel
-    var receive: ((Void throws -> [Int8]) -> Void)?
+    var channel: IOChannel!
+    var piped: Bool = false
 
     init(socket: TCPClientSocket) {
         self.socket = socket
@@ -38,25 +38,25 @@ final class TCPStream: StreamType {
             result.failure { error in
                 print(error)
             }
-            socket.close()
+            if !self.piped {
+                socket.close()
+            }
         }
 
         channel.setLowWater(1)
+    }
 
+    func receive(completion: (Void throws -> [Int8]) -> Void) {
         channel.read { result in
             result.success { done, data in
-                self.receive?({ data })
+                completion({ data })
             }
             result.failure { error in
-                self.receive?({ throw error })
+                completion({ throw error })
             }
         }
     }
 
-    func receive(receive: (Void throws -> [Int8]) -> Void) {
-        self.receive = receive
-    }
-
     func send(data: [Int8], completion: (Void throws -> Void) -> Void) {
         channel.write(data: data) { result in
             result.success { done, _ in
@@ -73,4 +73,10 @@ final class TCPStream: StreamType {
     func close() {
         channel.close()
     }
+    
+    func pipe() -> StreamType {
+        piped = true
+        channel.close()
+        return TCPStream(socket: socket)
+    }
 }

Aeon/HTTPServerType/HTTPServerType.swift

@@ -57,7 +57,7 @@ extension HTTPServerType {
                             self.serializer.serializeResponse(stream, response: response) { serializeResult in
                                 do {
                                     try serializeResult()
-                                    completion({ stream })
+                                    completion({ stream.pipe() })
                                 } catch {
                                     completion({ throw error })
                                     failure(error)