如何正确使用NSStream(无阻塞、读取到数据末尾以及消息重试)

Question

我正在尝试使用NSInputStream和NSOutputStream，但这会带来很多痛苦。

我有两个设备可以与Json通信。一些数据可能很长，因此NSOutputStreamsends将其拆分为多个数据包。

我需要在主线程上接收不阻塞的数据，并能够在尝试解析之前读取所有需要的json数据包。然后继续读取其余json数据数据包。

我需要发送不在主线程上阻塞，并且能够在第一批发送失败的情况下完成数据发送。然后继续发送其余的json数据。

我使用的是swift，但也可以使用目标c.

这是迄今为止的代码。我的基本流助手类：

public class StreamHelper : NSObject, NSStreamDelegate {
    static let DATA_BYTE_LENGTH = 4;
    public static func writeToOutputStream(text: String!, outputStream:NSOutputStream!) -> Int!{
        let encodedDataArray = [UInt8](text.utf8)
        var count: Int = encodedDataArray.count.littleEndian
        //convert int to pointer, which is required for the write method.
        withUnsafePointer(&count) { (pointer: UnsafePointer<Int>) -> Void in
            outputStream.write(UnsafePointer<UInt8>(pointer), maxLength: DATA_BYTE_LENGTH)
        }
        let bytesWritten = outputStream.write(encodedDataArray, maxLength: encodedDataArray.count)
        return bytesWritten;
    }
    public static func readFromInputStream(inputStream: NSInputStream!) -> String!{
        var buffer = [UInt8](count: 4096, repeatedValue: 0)
        var text = ""
        while (inputStream.hasBytesAvailable){
            let len = inputStream!.read(&buffer, maxLength: buffer.count)
            if(len > 0){
                if let output = NSString(bytes: &buffer, length: buffer.count, encoding: NSUTF8StringEncoding) as? String{
                    if (!output.isEmpty){
                        text += output
                    }
                }
            }
        }
        return text
    }
}

核心代码：

public func stream(aStream: NSStream, handleEvent eventCode: NSStreamEvent) {
    print("Reading from stream... ")
    switch (eventCode){
        case NSStreamEvent.ErrorOccurred:
            print("ErrorOccurred")
            break
        case NSStreamEvent.None:
            print("None")
            break
        case NSStreamEvent.EndEncountered:
            print("EndEncountered")
            if((aStream == inputStream) && inputStream!.hasBytesAvailable){
                // If all data hasn't been read, fall through to the "has bytes" event
            } else{
                break
            }
        case NSStreamEvent.HasBytesAvailable:
            print("HasBytesAvaible")
            let methodJson = StreamHelper.readFromInputStream(inputStream!)
            if(!methodJson.isEmpty){
                let cMethodJson = methodJson.cStringUsingEncoding(NSUTF8StringEncoding)!
                let returnedJsonString = String.fromCString(callMethod(cMethodJson))
                StreamHelper.writeToOutputStream(returnedJsonString, outputStream: outputStream!)
            }
            break
        case NSStreamEvent.OpenCompleted:
            print("OpenCompleted")
            break
        case NSStreamEvent.HasSpaceAvailable:
            print("HasSpaceAvailable")
            if(aStream == outputStream){
            }
            break
        default:
            break
    }
}

一些设置代码：

func connectToService(service: NSNetService!){
service.getInputStream(&inputStream, outputStream: &outputStream)
inputStream!.delegate = self
outputStream!.delegate = self
inputStream!.scheduleInRunLoop(NSRunLoop.currentRunLoop(), forMode: NSDefaultRunLoopMode)
outputStream!.scheduleInRunLoop(NSRunLoop.currentRunLoop(), forMode: NSDefaultRunLoopMode)
inputStream!.open()
outputStream!.open()
}

如何正确地使用NSStream，或者有比使用NSStream更好的解决方案吗？

Answer 1

你可能在这里工作太辛苦了。NSStreamDelegate是在GCD之前设计的，当时Cocoa的大部分工作都是在单个线程上完成的。虽然在某些情况下仍然有理由使用它，但在大多数情况下，GCD和同步方法会使它变得更容易。例如，为了阅读，你会做这样的事情：

import Foundation
enum StreamError: ErrorType {
    case Error(error: NSError?, partialData: [UInt8])
}
func readStream(inputStream: NSInputStream) throws -> [UInt8] {
    let bufferSize = 1024
    var buffer = [UInt8](count: bufferSize, repeatedValue: 0)
    var data: [UInt8] = []
    while true {
        let count = inputStream.read(&buffer, maxLength: buffer.capacity)
        guard count >= 0 else {
            inputStream.close()
            throw StreamError.Error(error: inputStream.streamError, partialData: data)
        }
        guard count != 0 else {
            inputStream.close()
            return data
        }
        data.appendContentsOf(buffer.prefix(count))
    }
}
let textPath = NSBundle.mainBundle().pathForResource("text.txt", ofType: nil)!
let inputStream = NSInputStream(fileAtPath: textPath)!
inputStream.open()
do {
    let data = try readStream(inputStream)
    print(data)
} catch let err {
    print("ERROR: (err)")
}

这肯定会阻塞当前队列。所以不要在主队列上运行它。将do块放入dispatch_async中。如果以后需要主队列上的数据，dispatch_async会将其返回，就像任何其他后台进程一样。

Answer 2

我不使用NSstream，而是直接通过套接字进行发送
有一些包装器可以让这项任务变得更容易，最著名的是GCDAsyncSocket。

Answer 3

基于其他答案，我提出了这个问题，它在Swift 4.2中有效。

public enum StreamError: Error {
    case Error(error: Error?, partialData: [UInt8])
}
extension InputStream {
    public func readData(bufferSize: Int = 1024) throws -> Data {
        var buffer = [UInt8](repeating: 0, count: bufferSize)
        var data: [UInt8] = []
        open()
        while true {
            let count = read(&buffer, maxLength: buffer.capacity)
            guard count >= 0 else {
                close()
                throw StreamError.Error(error: streamError, partialData: data)
            }
            guard count != 0 else {
                close()
                return Data(bytes: data)
            }
            data.append(contentsOf: (buffer.prefix(count)))
        }
    }
}