持续订阅和取消订阅 Observable 以进行 BLE 扫描

Question

我想实现一个功能，其中 BLE 扫描被激活 10 秒，然后中断 10 秒，并在视图处于活动状态时重复。

我正在使用 RxAndroidBle，它封装了 Android BLE API 以实现反应性。我有一个方法scanForRange(getConnectAddress());，它使用Observable<RxBleScanResult> 和一些过滤器和一个映射来返回一个简单的Observable<Integer>。要开始扫描，必须简单地订阅这个 observable，然后取消订阅以停止扫描。

我的第一个想法是使用Observable.interval()，如下所示：

@Override
protected void onResume() {
    super.onResume();

    mRangeScanSubscription = Observable.interval(10000, TimeUnit.MILLISECONDS)
            .flatMap(new Func1<Long, Observable<Integer>>() {
                @Override
                public Observable<Integer> call(Long aLong) {
                    return scanForRange(getConnectAddress());
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    Log.d(TAG, "onResume() error");
                    throwable.printStackTrace();
                }
            })
            .subscribe(new Action1<Integer>() {
                @Override
                public void call(Integer integer) {
                    Toast.makeText(DeviceDetailActivity.this, "Range: " + integer, Toast.LENGTH_SHORT).show();
                }
            });
}

然后我会在我的onPause() 中取消订阅该订阅。现在这不起作用，因为一旦发出第一个间隔，就会订阅扫描并将连续发出结果（而不是每 10 秒开始/停止）。

那么实现此功能的最佳方式是什么？我想我可能可以混合使用 Observables 和 Handlers 来破解它，但这似乎是错误的。

Answer 1

您可以使用TakeUntil 和RepeatWhen 运算符来做到这一点

        mRangeScanSubscription = scanForRange(getConnectAddress())
            .takeUntil(Observable.timer(10, TimeUnit.SECONDS)) //emits items for 10 seconds
            .repeatWhen(completed -> completed.delay(10, TimeUnit.SECONDS)) //resubscribes after 10 seconds
            ...