     }
   }
 
+  // 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>()
 

  */
  public var streamResponses = false

  private var socket: Socket?
  private var thread: Thread?

  public func run(at path: String) -> Completable {
    Completable.create { completable in
      self.queue.async {
        self.stopLock.withWriteLock {
          self.stopped = false

          do {
            try self.socket = Socket.create(
              family: .unix,
              type: .stream,
              proto: .unix
            )
            try self.socket?.connect(to: path)
            self.setUpThreadAndStartReading()
          } catch {
            self.streamSubject.onError(
              Error(msg: "Could not get socket", cause: error)
            )
            completable(.error(
              Error(msg: "Could not get socket at \(path)", cause: error)
            ))
          }
        }

        completable(.completed)

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

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