RxAndroidBle 2 在 Kotlin 中订阅了两个不同的特性答案

【问题标题】：RxAndroidBle 2 subscribe to two different characteristics in KotlinRxAndroidBle 2 在 Kotlin 中订阅了两个不同的特性
【发布时间】：2020-09-01 20:41:16
【问题描述】：

如果这是一个重复的问题，我深表歉意，我无法在任何地方找到有用的答案。我也是响应式编程的初学者，所以对于愚蠢的问题，我很抱歉。

我在 Kotlin Android 应用程序中使用 rxandroidble2。如果我只想读取一个特征，我可以成功连接到 BLE 设备。但是，我无法让它连接到两个独立的特征。

特征 A 每两秒发送一次数据包。特性 B 每秒发送数据包多次。如果我单独执行，我可以连接到它们两个，即运行应用程序时只为 CharA 或 CharB 建立连接。

我尝试过的解决方案都来自stackoverflow，特别是我尝试过：

我的代码是这样的：

respeckLiveSubscription = connectionObservable?.flatMap{
            Observable.combineLatest(
                it.setupNotification(UUID.fromString(Constants.CHARACTERISTIC_A_UUID)),
                it.setupNotification(UUID.fromString(Constants.CHARACTERISTIC_B_UUID)),
                io.reactivex.functions.BiFunction<Observable<ByteArray>, Observable<ByteArray>, Pair<Observable<ByteArray>, Observable<ByteArray>>> {
                    t1, t2 -> Pair.create(t1, t2)
                }
            )
        }?.subscribe({
            processRESpeckPacket(it.first.blockingFirst(), respeckVersion, this)
            Log.i("ble", "processed first packet")
            processRESpeckPacket(it.second.blockingFirst(), 7, this)
            Log.i("ble", "processed second packet")
        },{
            Log.i("ble", "error when connecting = " + it.stackTrace)
        })

        Log.i("ble", "does this mean connection is over")

我有几个问题：

subscribe 方法接收一个 Pair 类型的对象。当我尝试获取该对中的对象时，它们作为可观察对象返回，而我设法获取可观察值的唯一方法是调用blockingFirst()。显然这不是一个好的解决方案，因为我的代码无法运行。
我无法让 BiFunction 返回一对 ByteArrays 而不是 Observable。
如果我注释掉 processRESpeckPacket 行并留下日志，它们只会运行一次，这让我认为值更改通知不起作用。
我还在 Java 中尝试了上述大多数解决方案，但我得到了相同的行为。我链接到的 stackoverflow 答案中未解释的 rxandroidble2 是否发生了变化？
我的特征以不同的频率发送是否会导致所有问题？

请帮帮我，我一直被这个问题困扰！我怎样才能让它在 Kotlin 中工作？

非常感谢。

【问题讨论】：

标签： android kotlin bluetooth-lowenergy rxandroidble

【解决方案1】：

我的特征是以不同频率发送的事实是否会导致所有问题？

是的。从setupNotification() 发出的Observable<ByteArray> 是一个热门的Observable——不管你听与否，它都会发出——就像收音机一样。使用.blocking*() 操作符会阻塞你的监听线程，如果另一个Observable 在阻塞时间内发出，你会错过它。

我还在 Java 中尝试了上述大多数解决方案，并且得到了相同的行为。 rxandroidble2 中是否发生了我链接到的 stackoverflow 答案中未解释的更改？

RxJava 1 和 2 之间的更改主要是装饰性的，在适当的情况下从 Observable 更改为 Single 或 Completable。一些运营商已被删除或更改名称，但不是您的情况。

如果您只需要两个特征发出的最新值，您可以使用如下代码：

respeckLiveSubscription = connectionObservable?.flatMap{
            Observable.combineLatest(
                it.setupNotification(UUID.fromString(Constants.CHARACTERISTIC_A_UUID)).flatMap { it }, // notice flatMap
                it.setupNotification(UUID.fromString(Constants.CHARACTERISTIC_B_UUID)).flatMap { it }, // notice flatMap
                io.reactivex.functions.BiFunction<ByteArray, ByteArray, Pair<ByteArray, ByteArray>> {
                    t1, t2 -> Pair.create(t1, t2) // now we have 'ByteArray's here, not 'Observable<ByteArray>'s 
                }
            )
        }?.subscribe({
            processRESpeckPacket(it.first, respeckVersion, this)
            Log.i("ble", "processed first value")
            processRESpeckPacket(it.second, 7, this)
            Log.i("ble", "processed second value")
        },{
            Log.i("ble", "error when connecting = " + it.stackTrace)
        })

        Log.i("ble", "does this mean connection is over") // no it is only subscribed

【讨论】：