     }
   }
 
+  // MARK: - Private
   private var nextMsgid: UInt32 = 0
   private let nextMsgidLock = NSLock()
 
+  private var socket: Socket?
+  private var thread: Thread?
   private let queue = DispatchQueue(
     label: String(reflecting: RxMsgpackRpc.self),
     qos: .userInitiated
   )
 
   private var stopped = true
-  private let stopLock = ReadersWriterLock()
 
   private let streamSubject = PublishSubject<Message>()
 

          }
        } catch let error as Socket.Error {
          self.streamSubject.onError(Error(msg: "Could not read from socket", cause: error))
          // No need to lock since we are currently trying to open the socket.
          self.cleanUpAndCloseSocket()
          return
        } catch {
          self.streamSubject.onNext(

  public func stop() -> Completable {
    Completable.create { completable in
      self.queue.async {
        self.streamSubject.onCompleted()
        self.cleanUpAndCloseSocket()

        completable(.completed)
      }
      return Disposables.create()
    }
