                                            configuration.port,
                                            &readStream,
                                            &writeStream)
-        
-        input = readStream?.takeRetainedValue()
-        output = writeStream?.takeRetainedValue()
-        streamQueue.async { [weak self] in
-            
-            self?.configureAndOpen(stream: self?.input)
-            self?.configureAndOpen(stream: self?.output)
-            RunLoop.current.run()
-        }
-        streamQueue.async { [weak self] in
-            self?.startTimeoutTimer()
-        }
-    }
-    
-    private func configureAndOpen(stream: Stream?) {
-        
-        stream?.delegate = streamsDelegate
-        stream?.schedule(in: RunLoop.current, forMode: .defaultRunLoopMode)
-        stream?.open()
-    }
-    
+        guard let inputStream: InputStream = readStream?.takeRetainedValue(),
+            let outputStream: OutputStream = writeStream?.takeRetainedValue() else {
+            state = .error(.unknown)
+            return
+        }
+        pthread_rwlock_wrlock(&streamLock)
+        input = inputStream
+        output = outputStream
+        inputStream.delegate = self.streamsDelegate
+        outputStream.delegate = self.streamsDelegate
+        CFReadStreamSetDispatchQueue(input, streamQueue)
+        CFWriteStreamSetDispatchQueue(output, streamQueue)
+        outputStream.open()
+        outputStream.open()
+        pthread_rwlock_wrlock(&streamLock)
+    }
+
     // MARK: - Close -
     open func close() {
         guard state != .closed else { return }

    
    // MARK: - State handling & data received callback -
    /// The completion queue is used to dispatch this block.
    open var onStateChanged: OBDConnectionStateCallback? = nil
    open private(set) var state: OBDConnectionState = .closed {
        didSet { stateValueChanged(oldValue: oldValue) }
    }
    
    // MARK: - Streams -
    private var streamsDelegate: OBDStreamDelegate
    private var streamQueue = DispatchQueue(label: "OBDConnectionQueue", qos: .utility, attributes: .concurrent)
    private var input: InputStream?
    private var output: OutputStream?
    
    // MARK: - Request handling -
    private var requestResponse: String = ""
    private var requestTimeoutTimer: Timer?
    private var requestCompletion: OBDResultCallback?
    
    // MARK: - Init -
    public init(configuration: OBDConnectionConfiguration = OBDConnectionConfiguration.defaultELMAdapterConfiguration(),

        } else {
            error = .unknown
        }
        handle(connectionError: error)
    }
    
    private func handle(connectionError: OBDConnectionError) {
        
        flushConnection()
        state = .error(connectionError)
    }
    
    // MARK: - Timeout handling -
    private func startRequestTimeoutTimer() {
        
        invalidateRequestTimeoutTimer()
        let timer = DispatchSource.makeTimerSource(queue: streamQueue)
        timer.scheduleOneshot(deadline: .now() + configuration.requestTimeout)
        timer.setEventHandler { [weak self] in
            self?.finishTransmission(result: .failure(.requestTimeout))
        }
        timer.resume()
        requestTimeoutTimer? = timer
    }
    
    private func invalidateRequestTimeoutTimer() {
        
        requestTimeoutTimer?.cancel()
        requestTimeoutTimer = nil
    }
}
