           { _, msgid, data, info in
             guard let infoPtr = UnsafeRawPointer(info) else { return nil }
 
-            let handler = Unmanaged<MessageHandler>
-              .fromOpaque(infoPtr).takeUnretainedValue()
+            let handler = Unmanaged<MessageHandler>.fromOpaque(infoPtr).takeUnretainedValue()
             return handler.handleMessage(msgId: msgid, cfdata: data)
           },
           &localCtx,

    qos: .userInitiated
  )

  public init() {
  }

  public func send(
    msgid: Int32,
    data: Data?,
    expectsReply: Bool
  ) -> Single<Data?> {
    return Single.create { single in
      self.queue.async {
        self.portLock.withReadLock {
          guard self.portIsValid else {

  }

  public var syncReplyBody: SyncReplyBody? {
    get { self.messageHandler.syncReplyBody }
    set { self.messageHandler.syncReplyBody = newValue }
  }

  public var stream: Observable<Message> { self.streamSubject.asObservable() }

  public var queue = DispatchQueue(
    label: String(reflecting: RxMessagePortServer.self),
    qos: .userInitiated
  )

  public init() { self.messageHandler = MessageHandler(subject: self.streamSubject) }

  public func run(as name: String) -> Completable {
    Completable.create { completable in
      self.queue.async {
        var localCtx = CFMessagePortContext(
          version: 0,
