How do you correctly work with NSStreams (no blocking, read to the end of the data, and message retrying)?

I'm trying to work with NSInputStream and NSOutputStream but it's causing a lot of pain.

I have two devices that communicate Json. Some of the data can be very long so NSOutputStreamsends splits it up in multiple packets.

I need receiving not to block on the main thread and able to read all the needed json packets before trying to parse it. Then continue reading the rest json data packets.

I need sending not to block on the main thread and able to finish sending the data if the first batch failed to send. Then continue sending the rest of the json data.

I'm using swift but can also use objective c.

Here is the code so far. My basic stream helper class:

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
    }
}

Core code:

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
    }
}

Some setup code:

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()
}

How does one work with NSStreams correctly or is there a better solution than using NSStreams?

You're probably working too hard here. NSStreamDelegate was designed before GCD, when the majority of Cocoa work was done on a single thread. While there are still reasons to use it in some cases, in most cases GCD and synchronous methods will make it much easier. For example, to read you'd do something like this:

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)")
}

This will block the current queue, sure. So don't run it on the main queue. Put the do block into a dispatch_async. If you later need the data on the main queue, dispatch_async it back, just like any other background process.