RxAndroidBle:独立订阅者

Question

我是RxJava和RxAndroidBle的新手，如果您能帮助我解决正在尝试解决的问题，我将不胜感激。基本上，我有一个BLE设备，我订阅了四个特征。从这些可观察对象发出的数据被组合成一个可观察对象：

private RxBleDevice mBleDevice;
private Disposable mConnectionSubscription;

...

mConnectionSubscription = mBleDevice.establishConnection(false)
        .flatMap(rxBleConnection -> Observable.combineLatest(
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID0)).flatMap(observable -> observable),
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID1)).flatMap(observable -> observable),
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID2)).flatMap(observable -> observable),
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID3)).flatMap(observable -> observable),
                MyDataClass::new
        ))
        .observeOn(AndroidSchedulers.mainThread())
        .doFinally(this::disconnect)
        .subscribe(
                myData -> {
                    this.onNotificationReceived(myData);
                },
                this::onNotificationSetupFailure
        );

其中：

public class MyDataClass<Data0, Data1, Data2, Data3> {
    ...

    public MyDataClass(Data0 data0, Data1 data1, Data2 data2, Data3 data3) {
        ...
    }
}

上面的方法运行良好。我想要做的是订阅另一个特征。然而，这只是时不时地发出数据。因此，我想将这个额外的观察值连接到一个单独的观察者(比如onNotificationReceived2)。我不希望这些额外的可观察数据与MyDataClass相关。我该怎么做呢？

Answer 1

在您的flatMap()之前插入：.doOnNext( rxBleConnection -> savedConnection = rxBleConnection )，然后使用savedConnection设置您的其他订阅。

当您处理(或丢失)原始连接时，不要忘记将savedConnection设为空。

Answer 2

我找到了我想要的解决方案。根据this issue和Robert Lewis的评论，我最终做了以下事情：

private RxBleDevice mBleDevice;
private Disposable mConnectionSubscription1;
private Disposable mConnectionSubscription2;

...

Observable<RxBleConnection> sharedConnectionObservable = mBleDevice.establishConnection(false)
        .replay(1).refCount();

mConnectionSubscription1 = sharedConnectionObservable
        .flatMap(rxBleConnection -> Observable.combineLatest(
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID0)).flatMap(observable -> observable),
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID1)).flatMap(observable -> observable),
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID2)).flatMap(observable -> observable),
                rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID3)).flatMap(observable -> observable),
                MyDataClass::new
        ))
        .observeOn(AndroidSchedulers.mainThread())
        .doFinally(this::disconnect)
        .subscribe(
                myData -> {
                    this.onNotificationReceived(myData);
                },
                this::onNotificationSetupFailure
        );

mConnectionSubscription2 = sharedConnectionObservable
        .flatMap(rxBleConnection -> rxBleConnection.setupNotification(UUID.fromString(CHARACTERISTIC_UUID4)).flatMap(observable -> observable))
        .observeOn(AndroidSchedulers.mainThread())
        .doFinally(this::disconnect)
        .subscribe(
                myData2 -> {
                    this.onNotificationReceived2(myData2);
                },
                this::onNotificationSetupFailure
        );