setupNotification returns "Error already connected" 而没有发送连接请求

Question

我正在使用 RxAndroidBLE API 制作一个带有一些 BLE 交互的 Android 应用程序。我遵循了 https://github.com/Polidea/RxAndroidBle

中的示例指南和示例

我与指定设备建立了 BLE 连接，稍后在连接时我读写特性没有问题，但是当我尝试为电池电量特性设置通知时，我收到以下可抛出的错误消息：已经连接到地址为 MAC 的设备 XX:XX..."

我真的不明白那个上下文中的错误，因为我可以毫无问题地读写特性。

我想在出于特定目的初始读取其值后为该特征设置通知。

这是重现我的问题的示例代码：

private lateinit var device: RxBleDevice
private var connectionObservable: Observable<RxBleConnection>? = null
private var rxBleConnection: RxBleConnection? = null

private val connectionDisposable = CompositeDisposable()
private val connectionStateDisposable = CompositeDisposable()
private var notifyValueChangeSubscription = CompositeDisposable()


var enableBatteryNotificationRunnable: Runnable = Runnable {
  enableBatteryNotification()
}
private var myHandler = Handler()
val DELAY_BEFORE_ENABLE_NOTIFICATION: Long = 100


private fun connect() {
  connectionObservable = device.establishConnection(false)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())

  connectionObservable?.let {
    connectionDisposable.add(it.subscribe(
      { rxBleConnection ->
        this.rxBleConnection = rxBleConnection
      },
      { _ ->
        Log.e("connect", "connexion error")                       
      })
    )
  }

  val state = device.observeConnectionStateChanges().subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
  connectionStateDisposable.add(
    state.subscribe(
      { connectionState ->
        Log.i("connect", "connexion state :$connectionState")
        if(connectionState == RxBleConnection.RxBleConnectionState.CONNECTED) {
            myHandler.postDelayed(enableBatteryNotificationRunnable, DELAY_BEFORE_ENABLE_NOTIFICATION);
        }
      }
    )
    { _ ->
      Log.e("connection listener", "connexion state error")
    }
  )
}

private fun enableBatteryNotification () {
  connectionObservable?.let {
    var observableToReturn =  it
      .flatMap { it.setupNotification(UUID_BATTERY_LEVEL) }
      .doOnNext {
        Log.i("NOTIFICATION", "doOnNext")
      }
      .flatMap { it }
      .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())

    notifyValueChangeSubscription.add(observableToReturn.subscribe({ bytes ->
        var strBytes = String(bytes)
        Log.i("NOTIFICATION", "value change: $strBytes")
      }, 
      { throwable ->
        Log.e("NOTIFICATION", "Error in notification process: " + throwable.message)
      })
    )
  }
}

在此先感谢您的帮助:)

Answer 1

setupNotification returns “Error already connected” whereas no connection request is send

实际上发出了两个连接请求——因此出现错误。来自 RxBleDevice.establishConnection() Javadoc：

     * Establishes connection with a given BLE device. {@link RxBleConnection} is a handle, used to process BLE operations with a connected
     * device.

在您的代码中有两个订阅 establishConnection() Observable。

private lateinit var device: RxBleDevice
private var connectionObservable: Observable<RxBleConnection>? = null
private var rxBleConnection: RxBleConnection? = null

private val connectionDisposable = CompositeDisposable()
private val connectionStateDisposable = CompositeDisposable()
private var notifyValueChangeSubscription = CompositeDisposable()


var enableBatteryNotificationRunnable: Runnable = Runnable {
  enableBatteryNotification()
}
private var myHandler = Handler()
val DELAY_BEFORE_ENABLE_NOTIFICATION: Long = 100


private fun connect() {
  connectionObservable = device.establishConnection(false)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())

  connectionObservable?.let {
    connectionDisposable.add(it.subscribe( // << Here is the first subscription
      { rxBleConnection ->
        this.rxBleConnection = rxBleConnection
      },
      { _ ->
        Log.e("connect", "connexion error")                       
      })
    )
  }

  val state = device.observeConnectionStateChanges().subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
  connectionStateDisposable.add(
    state.subscribe(
      { connectionState ->
        Log.i("connect", "connexion state :$connectionState")
        if(connectionState == RxBleConnection.RxBleConnectionState.CONNECTED) {
            myHandler.postDelayed(enableBatteryNotificationRunnable, DELAY_BEFORE_ENABLE_NOTIFICATION);
        }
      }
    )
    { _ ->
      Log.e("connection listener", "connexion state error")
    }
  )
}

private fun enableBatteryNotification () {
  connectionObservable?.let {
    var observableToReturn =  it
      .flatMap { it.setupNotification(UUID_BATTERY_LEVEL) }
      .doOnNext {
        Log.i("NOTIFICATION", "doOnNext")
      }
      .flatMap { it }
      .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())

    notifyValueChangeSubscription.add(observableToReturn.subscribe({ bytes -> // << Here is the second subscription
        var strBytes = String(bytes)
        Log.i("NOTIFICATION", "value change: $strBytes")
      }, 
      { throwable ->
        Log.e("NOTIFICATION", "Error in notification process: " + throwable.message)
      })
    )
  }
}

这种情况是人们学习的常见困惑来源 RxJava。有 three paths 可以解决您的问题。工作量从最少到最多：

分享`establishConnectionObservable`

可以与 RxReplayingShare 共享一个 RxBleConnection。改变这个：

  connectionObservable = device.establishConnection(false)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())

为此：

  connectionObservable = device.establishConnection(false)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .compose(ReplayingShare.instance())

使用 `rxBleConnection: RxBleConnection?` 属性

而不是：

  connectionObservable?.let {
    var observableToReturn =  it
      .flatMap { it.setupNotification(UUID_BATTERY_LEVEL) }
      .doOnNext {
        Log.i("NOTIFICATION", "doOnNext")
      }
      .flatMap { it }
      .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())

    notifyValueChangeSubscription.add(observableToReturn.subscribe({ bytes -> // << Here is the second subscription
        var strBytes = String(bytes)
        Log.i("NOTIFICATION", "value change: $strBytes")
      }, 
      { throwable ->
        Log.e("NOTIFICATION", "Error in notification process: " + throwable.message)
      })
    )
  }

成功：

  rxBleConnection?.let {
    var observableToReturn = rxBleConnection.setupNotification(UUID_BATTERY_LEVEL)
      .doOnNext {
        Log.i("NOTIFICATION", "doOnNext")
      }
      .flatMap { it }
      .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())

    notifyValueChangeSubscription.add(observableToReturn.subscribe({ bytes -> // << Here is the second subscription
        var strBytes = String(bytes)
        Log.i("NOTIFICATION", "value change: $strBytes")
      }, 
      { throwable ->
        Log.e("NOTIFICATION", "Error in notification process: " + throwable.message)
      })
    )
  }

不鼓励这样做，因为您可能会得到一个不再有效的 RxBleConnection，因为它可能在调用 enableBatteryNotification()

之前已断开连接

更改代码流以使用单个 `.subscribe()`

这是根据您的具体用例量身定制的定制解决方案。不幸的是，您添加的信息不足以创建插入式代码替换，但它可能看起来像这样：

device.establishConnection(false)
    .flatMap { connection ->
        Observable.merge(
            connection.readCharacteristic(uuid0).map { ReadResult(uuid0, it) }.toObservable(),
            connection.setupNotification(uuid1).flatMap { it }.map { NotifyResult(uuid1, it) }.delaySubscription(100, TimeUnit.MILLISECONDS)
        )
    }
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        { /* handle ReadResult/NotifyResult */ },
        { /* handle potential errors */ }
    )

其中 ReadResult 和 NotifyResult 将是 data class 需要 UUID 和 ByteArray

setupNotification returns "Error already connected" 而没有发送连接请求

setupNotification returns "Error already connected" whereas no connection request is send

android

kotlin

bluetooth-lowenergy

rxandroidble

分享`establishConnectionObservable`

使用 `rxBleConnection: RxBleConnection?` 属性

更改代码流以使用单个 `.subscribe()`

setupNotification returns "Error already connected" 而没有发送连接请求

setupNotification returns "Error already connected" whereas no connection request is send

android

kotlin

bluetooth-lowenergy

rxandroidble

分享establishConnectionObservable

使用 rxBleConnection: RxBleConnection? 属性

更改代码流以使用单个 .subscribe()

分享`establishConnectionObservable`

使用 `rxBleConnection: RxBleConnection?` 属性

更改代码流以使用单个 `.subscribe()`