Tengo un servidor que me envía mensajes a través de TCP donde los primeros 4 bytes determinan la longitud del resto del mensaje. Entonces necesito

1) leer 4 bytes en un UInt32 (funciona) y almacenarlo enbytes_esperados

2) leerbytes_esperados bytes enmensaje

En este momento mi código se ve así:

private let inputStreamAccessQueue  = DispatchQueue(label: "SynchronizedInputStreamAccess")

func inputStreamHandler(_ event: Stream.Event) {
    switch event {
        case Stream.Event.hasBytesAvailable:
            self.handleInput()

        ...
    }
}

func handleInput() {
    // **QUESTION: Do I use this barrier wrong?**
    self.inputStreamAccessQueue.sync(flags: .barrier) {            
        guard let istr = self.inputStream else {
            log.error(self.buildLogMessage("InputStream is nil"))
            return
        }

        guard istr.hasBytesAvailable else {
            log.error(self.buildLogMessage("handleInput() called when inputstream has no bytes available"))
            return
        }

        let lengthbuffer = UnsafeMutablePointer<UInt8>.allocate(capacity: 4)
        defer { lengthbuffer.deallocate(capacity: 4) }
        let lenbytes_read = istr.read(lengthbuffer, maxLength: 4)

        guard lenbytes_read == 4 else {
            self.errorHandler(NetworkingError.InputError("Input Stream received \(lenbytes_read) (!=4) bytes"))
            return
        }

        let bytes_expected = Int(UnsafeRawPointer(lengthbuffer).load(as: UInt32.self).bigEndian)
        log.info(self.buildLogMessage("expect \(bytes_expected) bytes"))

        print("::DEBUG", call, "bytes_expected", bytes_expected)

        var message = ""
        var bytes_missing = bytes_expected
        while bytes_missing > 0 {
            //print("::DEBUG", call, "bytes_missing", bytes_missing)
            let buffer = UnsafeMutablePointer<UInt8>.allocate(capacity: bytes_missing)
            let bytes_read = istr.read(buffer, maxLength: bytes_missing)

            print("::DEBUG", call, "bytes_read", bytes_read)

            guard bytes_read > 0 else {
                print("bytes_read not > 0: \(bytes_read)")
                return
            }

            guard bytes_read <= bytes_missing else {
                print("Read more bytes than expected. missing=\(bytes_missing), read=\(bytes_read)")
                return
            }

            guard let partial_message = String(bytesNoCopy: buffer, length: bytes_read, encoding: .utf8, freeWhenDone: true) else {
                log.error("ERROR WHEN READING")
                return
            }

            message = message + partial_message
            bytes_missing -= bytes_read
        }

        self.handleMessage(message)
    }
}

Mi problema es que istr.read (buffer, maxLength: bytes_missing) a veces no lee todos los mensajes a la vez, por lo que hago un bucle hasta que haya leído todo lo que quiero. Pero todavía veo que mi aplicación falla (raramente) porque se vuelve a llamar a handleInput () mientras aún se está ejecutando otra llamada a ese método. En este caso, bytes_esperados contiene valores aleatorios y la aplicación se bloquea debido a la asignación ilegal de memoria.

Pensé que podría evitar esto usando la barrera. Pero parece que esto no funciona ... ¿Estoy usando mal la barrera?