             return callback(.Error(SuvError.RuntimeError(message: "Stream is already closed")))
         }
         
-        onRead2 = { result in
+        let onRead2: GenericResult<Pipe> -> () = { result in
             if case .Success(let queue) = result {
                 let pipePtr = UnsafeMutablePointer<uv_pipe_t>(queue.streamPtr)
                 if uv_pipe_pending_count(pipePtr) <= 0 {

        }
        
        var data = uv_buf_init(bytes, length)
        self.onWrite = onWrite
        
        withUnsafePointer(&data) {
            let writeReq = UnsafeMutablePointer<uv_write_t>(allocatingCapacity: sizeof(uv_write_t))
            writeReq.pointee.data = unsafeBitCast(self, to: UnsafeMutablePointer<Void>.self)
            
            let r = uv_write(writeReq, streamPtr, $0, 1) { req, _ in
                let stream = unsafeBitCast(req.pointee.data, to: Stream.self)
                destroy_write_req(req)
                stream.onWrite(.Success)
            }
            
            if r < 0 {

            return callback(.Error(SuvError.RuntimeError(message: "Stream is already closed")))
        }
        
        streamPtr.pointee.data = retainedVoidPointer(callback)
        
        let r = uv_read_start(streamPtr, alloc_buffer) { stream, nread, buf in
            defer {
                dealloc(buf.pointee.base, capacity: nread)
            }
            
            let onRead: ReadStreamResult -> () = releaseVoidPointer(stream.pointee.data)!
            
            let data: ReadStreamResult
            if (nread == Int(UV_EOF.rawValue)) {
