你如何正确使用NSStreams（没有阻塞，读到数据末尾和消息重试）？

Question

我正在尝试与NSInputStream和NSOutputStream合作，但它造成了很多痛苦。

我有两个与Json通信的设备。有些数据可能很长，所以NSOutputStreamsends将它分成多个数据包。

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

我需要发送不阻塞主线程并能够在第一批发送失败时完成发送数据。然后继续发送剩余的json数据。

我正在使用swift但也可以使用Objective 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()
}

如何正确使用NSStreams或者是否有比使用NSStreams更好的解决方案？

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

我没有使用NSStreams，而是直接通过套接字发送。 有一些包装器可以使这项任务更容易，最明显的是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)))
        }

    }
}