RxJava испускается только тогда, когда эмиттер останавливается

Question

В Android-приложении я хотел бы обновить список только после прекращения выбора списка элементов в списке. Поэтому, по сути, я хотел бы, чтобы наблюдатель был проинформирован только после того, как производитель прекратил излучать не менее 500 мс.

Сейчас у меня есть что-то вроде следующего:

Subject<Object> _bus = PublishSubject.create().toSerialized(); 

... 
_bus.onNext(new Event()); 
... 

_bus.delay(500, TimeUnit.MILLISECONDS) 
    .distinctUntilChanged() 
    .observeOn(Schedulers.computation()) 
    .subscribe(event -> { 
     // Do something with event 
    })); 

Это нормально, за исключением того, что излучает на 500 мс с интервалом, даже если источник еще излучающие. Я хотел бы подождать 500 мс, чтобы узнать, не прекратил ли источник звонить onNext() и только потом испускает.

Возможно ли это?

Answer 1

Таким образом, в основном вам требуется дебуляция с буфером. Существует article, который должен помочь вам.

И пнуть образец из этой статьи:

Observable<Object> tapEventEmitter = _rxBus.toObserverable().share(); 
    Observable<Object> debouncedEventEmitter = tapEventEmitter.debounce(1, TimeUnit.SECONDS); 
    Observable<List<Object>> debouncedBufferEmitter = tapEventEmitter.buffer(debouncedEventEmitter); 

    debouncedBufferEmitter.buffer(debouncedEventEmitter) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Action1<List<Object>>() { 
      @Override 
      public void call(List<Object> taps) { 
      _showTapCount(taps.size()); 
      } 
     }); 

Answer 2

Я думаю, вы должны использовать оператор debounce вместо задержки, например.

_bus.debounce(500, TimeUnit.MILLISECONDS 
.distinctUntilChanged() 
.observeOn(Schedulers.computation()) 
.subscribe(event -> { 
    // Do something with event 
}));