 
     // MARK: - Keep alive timer
 
-    fileprivate func delayedPing(interval: UInt16) {
+    private func delayPing(interval: UInt16) {
         let time = DispatchTime.now() + Double(interval / 2)
         DispatchQueue.main.asyncAfter(deadline: time) { [weak self] in
-            guard
-                let _ = self?.outputStream
-                else { return }
-
             self?.mqttPing()
-            self?.delayedPing(interval: interval)
+            self?.delayPing(interval: interval)
         }
     }
 
     // MARK: - Socket connection
 
-    fileprivate func socketConnect(host: String, port: Int) -> Bool {
+    private func socketConnect(host: String, port: Int) -> Bool {
         Stream.getStreamsToHost(withName: host, port: port, inputStream: &inputStream, outputStream: &outputStream)
 
-        if inputStream == nil || outputStream == nil {
+        guard let input = inputStream, let output = outputStream else {
             return false
         }
 
-        serialQueue = DispatchQueue(label: "readQueue", qos: .background, target: nil)
-
-        inputStream?.delegate = self
-        outputStream?.delegate = self
-
-        inputStream?.schedule(in: .current, forMode: .defaultRunLoopMode)
-        outputStream?.schedule(in: .current, forMode: .defaultRunLoopMode)
-
-        inputStream?.open()
-        outputStream?.open()
+        input.open()
+        output.open()
+
+        DispatchQueue.global(qos: .background).async {
+            self.readStream(inputStream: input)
+        }
 
         return true
     }
 
-    fileprivate func socketDisconnect() {
-        inputStream?.remove(from: .current, forMode: .defaultRunLoopMode)
-        outputStream?.remove(from: .current, forMode: .defaultRunLoopMode)
-
-        inputStream?.delegate = nil
-        outputStream?.delegate = nil
-
+    private func socketDisconnect() {
         inputStream?.close()
         outputStream?.close()
 
         inputStream = nil
         outputStream = nil
-
-        serialQueue = nil
-    }
-
-    // MARK: - Stream delegate
-
-    @objc internal func stream(_ stream: Stream, handle eventCode: Stream.Event) {
-        switch stream {
-        case let value where value == inputStream:
-            switch (eventCode) {
-            case Stream.Event.hasBytesAvailable:
-                serialQueue?.async { [weak self] in
-                    self?.readStream()
-                }
-
-            case Stream.Event.openCompleted:
-                messageParserState = .decodingHeader
-            case Stream.Event.errorOccurred:
-                disconnect()
-            case Stream.Event.endEncountered:
-                disconnect()
-            default:
-                break
-            }
-
-        default:
-            break
-        }
-    }
-
-    fileprivate func readStream() {
-        guard
-            let inputStream = inputStream, inputStream.streamStatus == .open
-            else { return }
-
-        if messageParserState == .decodingHeader {
-            let count = inputStream.read(messageBuffer, maxLength: 1) // returns -1 count sometimes during active mqtt connection
-            if count > 0 {
+    }
+
+    // MARK: - Stream reading
+
+    private func readStream(inputStream: InputStream) {
+        let messageBuffer = UnsafeMutablePointer<UInt8>.allocate(capacity: LightMQTT.BUFFER_SIZE)
+        var byteCount = 0
+
+        var messageParserState: MQTTMessageParserState = .decodingHeader
+        var messageType: MQTTMessage = .connack
+
+        var messageLengthMultiplier = 1
+        var messageLength = 0
+
+        while inputStream.streamStatus == .opening {
+            usleep(1000)
+        }
+
+        while inputStream.streamStatus == .open {
+            messageLengthMultiplier = 1
+            messageLength = 0
+            byteCount = 0
+
+            while messageParserState == .decodingHeader && inputStream.streamStatus == .open {
+                let count = inputStream.read(messageBuffer, maxLength: 1)
+                if count == 0 {
+                    break
+                } else if count < 0 {
+                    return
+                }
+
                 if let message = MQTTMessage(rawValue: messageBuffer.pointee & 0xf0) {
                     messageType = message
                     messageParserState = .decodingLength
                 }
             }
-        }
-
-        while messageParserState == .decodingLength {
-            let count = inputStream.read(messageBuffer, maxLength: 1)
-            if count > 0 {
+
+            while messageParserState == .decodingLength && inputStream.streamStatus == .open {
+                let count = inputStream.read(messageBuffer, maxLength: 1)
+                if count == 0 {
+                    break
+                } else if count < 0 {
+                    return
+                }
+
                 messageLength += Int(messageBuffer.pointee & 127) * messageLengthMultiplier
                 if messageBuffer.pointee & 128 == 0x00 {
                     messageParserState = .decodingData
                 } else {
                     messageLengthMultiplier *= 128
                 }
-
-            } else if count < 0 {
-                disconnect()
-                return
-
-            } else {
-                break
             }
-        }
-
-        if messageParserState == .decodingData {
-            switch messageType {
-            case .publish:
-                guard messageLength > 0 else {
-                    resetParser()
+
+            while messageParserState == .decodingData && inputStream.streamStatus == .open {
+                switch messageType {
+                case .publish:
+                    guard messageLength > 0 else {
+                        messageParserState = .decodingHeader
+                        break
+                    }
+
+                    let bytesToRead = min(LightMQTT.BUFFER_SIZE, messageLength - byteCount)
+                    let count = inputStream.read(messageBuffer.advanced(by: byteCount), maxLength: bytesToRead)
+                    if count == 0 {
+                        break
+                    } else if count < 0 {
+                        return
+                    }
+
+                    byteCount += count
+
+                    if byteCount == messageLength {
+                        let topicLength = Int(messageBuffer.pointee) * 256 + Int(messageBuffer.advanced(by: 1).pointee)
+                        guard byteCount > topicLength + 2 else {
+                            messageParserState = .decodingHeader
+                            break
+                        }
+
+                        let topicPointer = UnsafeBufferPointer(start: messageBuffer + 2, count: topicLength)
+                        guard let topic = String(bytes: topicPointer, encoding: String.Encoding.utf8) else {
+                            messageParserState = .decodingHeader
+                            break
+                        }
+
+                        let pointer = UnsafeBufferPointer(start: messageBuffer + topicLength + 2,
+                                                          count: byteCount - topicLength - 2)
+
+                        if let closure = receivingMessage, let message = String(bytes: pointer, encoding: String.Encoding.utf8) {
+                            closure(topic, message)
+                        }
+
+                        if let closure = receivingBuffer {
+                            closure(topic, pointer)
+                        }
+
+                        if let closure = receivingBytes {
+                            closure(topic, Array(pointer))
+                        }
+
+                        if let closure = receivingData {
+                            closure(topic, Data(bytesNoCopy: messageBuffer + topicLength + 2,
+                                                count: byteCount - topicLength - 2,
+                                                deallocator: .none))
+                        }
+
+                        messageParserState = .decodingHeader
+                    }
+                    
+                case .pingresp:
+                    messageParserState = .decodingHeader
+
+                case .suback:
+                    messageParserState = .decodingHeader
+
+                case .connack:
+                    messageParserState = .decodingHeader
+
+                case .unsuback:
+                    messageParserState = .decodingHeader
+
+                case .disconnect:
                     return
-                }
-
-                let bytesToRead = min(LightMQTT.BUFFER_SIZE, messageLength - byteCount)
-                let count = inputStream.read(messageBuffer.advanced(by: byteCount), maxLength: bytesToRead)
-                if count < 0 {
-                    disconnect()
-                    return
-                }
-
-                byteCount += count
-
-                if byteCount == messageLength {
-                    let topicLength = Int(messageBuffer.pointee) * 256 + Int(messageBuffer.advanced(by: 1).pointee)
-                    let topicPointer = UnsafeBufferPointer(start: messageBuffer + 2, count: topicLength)
-
-                    guard
-                        let topic = String(bytes: topicPointer, encoding: String.Encoding.utf8)
-                        else { break }
-
-                    let pointer = UnsafeBufferPointer(start: messageBuffer + topicLength + 2,
-                                                      count: byteCount - topicLength - 2)
-
-                    if  let closure = receivingMessage,
-                        let message = String(bytes: pointer, encoding: String.Encoding.utf8) {
-                        closure(topic, message)
-                    }
-
-                    if let closure = receivingBuffer {
-                        closure(topic, pointer)
-                    }
-
-                    if let closure = receivingBytes {
-                        closure(topic, Array(pointer))
-                    }
-
-                    if let closure = receivingData {
-                        closure(topic, Data(bytesNoCopy: messageBuffer + topicLength + 2,
-                                            count: byteCount - topicLength - 2,
-                                            deallocator: .none))
-                    }
-
-                    resetParser()
-                }
-
-            case .pingresp:
-                resetParser()
-
-            case .suback:
-                resetParser()
-
-            case .connack:
-                resetParser()
-
-            case .unsuback:
-                resetParser()
-
-            case .disconnect:
-                disconnect()
-
-            default:
-                break
+
+                default:
+                    messageParserState = .decodingHeader
+                }
             }
         }
-    }
-
-    fileprivate func resetParser() {
-        byteCount = 0
-        messageLength = 0
-        messageLengthMultiplier = 1
-
-        messageParserState = .decodingHeader
-        messageType = .connack
+
+        messageBuffer.deinitialize(count: LightMQTT.BUFFER_SIZE)
+        messageBuffer.deallocate(capacity: LightMQTT.BUFFER_SIZE)
     }
 
     // MARK: - MQTT messages

        outputStream?.write(messageBytes, maxLength: messageBytes.count)
    }

    fileprivate func mqttDisconnect() {
        let messageBytes: [UInt8] = [
            0xe0,                               // FIXED BYTE 1   e = DISCONNECT, 0 = DUP QoS RETAIN (not used)
            0x00                                // FIXED BYTE 2   remaining length = 0

     * |--------------------------------------
     */

    private func mqttConnect(keepalive: UInt16) {
        let baseIntA = Int(arc4random() % 65535)
        let baseIntB = Int(arc4random() % 65535)
        let client = "client_" + String(format: "%04X%04X", baseIntA, baseIntB)
