Непрерывная подписка и отмена подписки на Наблюдение для сканирования BLE

Question

Я хотел бы реализовать функцию, в которой сканирование BLE активируется в течение 10 секунд, затем перерывы на 10 секунд и повторяется, пока объект активен.

Я использую RxAndroidBle, который обертывает API Android BLE для реактивности. У меня есть метод scanForRange(getConnectAddress());, который использует Observable<RxBleScanResult> и некоторые фильтры и карту, применяемую для возврата простого Observable<Integer>. Чтобы начать сканирование, этот наблюдаемый должен просто быть подписан, а затем отменен, чтобы остановить сканирование.

Моя первая мысль была использовать 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 
      ...