android - 在订阅 observable 之后执行操作

Question

我想执行一个操作，该操作在订阅 Observable 后 执行。目前我执行一些 BLE 相关操作，比如读取我的连接的 rssi:

public class BleGattOperations {

   private class BleGattCallback extends BluetoothGattCallback {
       public void onReadRemoteRssi(BluetoothGatt gatt, int rssi, int status) {
           readRssiSubject.onNext(new RssiState(gatt, rssi, status));
       }
   }

    // Defined in Constructor
    private final BluetoothGattCallback callback;
    // Defined in Constructor
    private BluetoothGatt gatt;
    private final Subject<RssiState, RssiState> readRssiSubject = PublishSubject.create();

    public Observable<RssiState> readRssi() {
        return readRssiSubject
            .doOnSubscribe(() -> {
                 gatt.readRemoteRssi();
            })
            .take(1);
    }
}

在 readRssi() 方法中，我订阅了 readRssiSubject，它在 BleGattCallback.onReadRemoteRssi-Method() 中发出项目。 ble 端的读取 rssi 操作是通过 doOnSubscribe() block 内的 gatt.readRemoteRssi() 操作触发的。

不幸的是，doOnSubscribe 方法是在之前执行任何对主题的订阅，有时读取 rssi 操作的执行速度比对 readRssiSubject 的订阅完成。因此，在执行 onReadRemoteRssi 方法后，新订阅的订阅者不会在订阅完成时收到 rssi 测量的结果。

也许一些实验代码 fragment 有助于理解基本上发生了什么:

public Observable<RssiState> readRssi() {
    return readRssiSubject
        .doOnSubscribe(() -> {
             readRssiSubject.onNext(new RssiState());
        })
        .take(1);
}

是否有一种方法可以在对 Subject/Observable 进行订阅后执行操作？

Answer 1

来自 doOnSubscribe 文档:

Each subscription will result in an invocation of the given action except when the source ObservableSource is reference counted, in which case the source ObservableSource will invoke the given action for the first subscription.

PublishSubject 是引用计数的。

替代方案

尽管您正在使用 PublishSubject，但您似乎并不打算将来自它的消息分享给多个订阅者。在这种情况下，我会推荐 readRssi() 的以下实现:

public Observable<RssiState> readRssi() {
    return Observable.create(emitter -> {
             // this is invoked for every new subscriber
             BluetoothGattCallback callback = 
                 (BluetoothGatt gatt, int rssi, int status) -> {
                     emitter.onNext(new RssiState(gatt, rssi, status));
                     emitter.onComplete();
                 };
             gatt.addListener(callback); // not sure about method name
             gatt.readRemoteRssi(); // initiate read
        });
}

参见 documentation and example对于 create 方法。