             output.setProperty(StreamSocketSecurityLevel.tlSv1, forKey: .socketSecurityLevelKey)
         }
 
-        input.open()
-        output.open()
-
-        while input.streamStatus == .opening || output.streamStatus == .opening {
-            usleep(1000)
-        }
-
-        if input.streamStatus != .open || output.streamStatus != .open {
-            return nil
-        }
-
-        DispatchQueue.global(qos: options.readQosClass).async {
-            self.readStream(input: input, output: output)
-        }
-
-        return (input, output)
+        DispatchQueue.global(qos: .userInitiated).async {
+            input.open()
+            output.open()
+
+            while input.streamStatus == .opening || output.streamStatus == .opening {
+                usleep(1000)
+            }
+
+            if input.streamStatus != .open || output.streamStatus != .open {
+                completion(nil)
+            }
+            
+            completion((input, output))
+        }
     }
 
     private func closeStreams() {

        remainingLengthBytes                    // remaining length, variable

        let messageBytes = headerBytes + encode(string: topic) + [UInt8](message)
        outputStream?.write(messageBytes, maxLength: messageBytes.count)
    }

    @objc private func mqttPing(outputStream: OutputStream) {
        let messageBytes: [UInt8] = [
            0xc0,                               // FIXED BYTE 1   c = PINGREQ, 0 = DUP QoS RETAIN (not used)
            0x00                                // FIXED BYTE 2   remaining length = 0
        ]
        
        outputStream.write(messageBytes, maxLength: messageBytes.count)
    }
    
    // MARK: - Utils

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

    private func mqttConnect(output: OutputStream, keepalive: UInt16) {
        /**
         * |----------------------------------------------------------------------------------
         * |     7    |    6     |      5     |  4   3  |     2    |       1      |     0    |
