RXJava – hacer un observable pausable (con buffer y ventana, por ejemplo)

Quiero crear observables que hagan lo siguiente:

• Almacenar todos los elementos, mientras están en pausa
• Inmediatamente emiten elementos, mientras que no se detienen
• El gatillo de pausa / reanudación debe provenir de otro observable
• Debe ser guardar para ser utilizado por observables que no se ejecutan en el hilo principal y debe ser guardar cambiar el estado de pausa / reanudación del hilo principal

Quiero usar un BehaviorSubject<Boolean> como desencadenador y enlazar este desencadenador al evento onResume y onPause una actividad. (Ejemplo de código adjunto)

• Comprobación de Internet, dónde colocar al usar MVP, RX y Retrofit
• ¿Dónde se puede encontrar AndroidObservable.fromBroadcast ahora?
• Encadenamiento de servicios de Retrofit con soporte de RxJava
• RxJava - Encadenamiento de solicitudes y actualización de la interfaz de usuario
• Robolectric Test no llama a textWatcher.onTextChanged

Pregunta

He configurado algo, pero no está funcionando como estaba previsto. Lo uso como sigue:

 Observable o = ...; // Variant 1 o = o.lift(new RxValve(getPauser(), 1000, getPauser().getValue()) // Variant 2 // o = o.compose(RXPauser.applyPauser(getPauser())); o .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(); 

Actualmente el problema es que la Variante 1 debería funcionar bien, pero a veces, los eventos no son emitidos – la válvula no emite, hasta que la válvula todo funcione (puede ser un problema de roscado …)! La solución 2 es mucho más simple y parece funcionar, pero no estoy seguro si es realmente mejor, no lo creo. En realidad no estoy seguro, ¿por qué una solución es fallar a veces, así que no estoy seguro si la solución 2 resuelve el problema (actualmente para mí desconocido) …

¿Puede alguien decirme cuál podría ser el problema o si la solución simple debe trabajar confiablemente? ¿O puede mostrarme una solución confiable?

Código

RxValue

Https://gist.github.com/akarnokd/1c54e5a4f64f9b1e46bdcf62b4222f08

Funciones de RXPauser

 public static <T> Observable.Transformer<T, T> applyPauser(Observable<Boolean> pauser) { return observable -> pauser(observable, pauser); } private static <T> Observable<T> pauser(Observable<T> source, Observable<Boolean> pauser) { // this observable buffers all items that are emitted while emission is paused Observable<T> sharedSource = source.publish().refCount(); Observable<T> queue = sharedSource .buffer(pauser.distinctUntilChanged().filter(isResumed -> !isResumed), aBoolean -> pauser.distinctUntilChanged().filter(isResumed -> isResumed)) .flatMap(l -> Observable.from(l)) .doOnNext(t -> Ld(RXPauser.class, "Pauser QUEUED: " + t)); // this observable emits all items that are emitted while emission is not paused Observable<T> window = sharedSource.window(pauser.distinctUntilChanged().filter(isResumed -> isResumed), aBoolean -> pauser.distinctUntilChanged().filter(isResumed -> !isResumed)) .switchMap(tObservable -> tObservable) .doOnNext(t -> Ld(RXPauser.class, "Pauser NOT QUEUED: " + t)); // combine both observables return queue.mergeWith(window) .doOnNext(t -> Ld(RXPauser.class, "Pauser DELIVERED: " + t)); } 

Actividad

 public class BaseActivity extends AppCompatActivity { private final BehaviorSubject<Boolean> pauser = BehaviorSubject.create(false); public BaseActivity(Bundle savedInstanceState) { super(args); final Class<?> clazz = this.getClass(); pauser .doOnUnsubscribe(() -> { Ld(clazz, "Pauser unsubscribed!"); }) .subscribe(aBoolean -> { Ld(clazz, "Pauser - " + (aBoolean ? "RESUMED" : "PAUSED")); }); } public PublishSubject<Boolean> getPauser() { return pauser; } @Override protected void onResume() { super.onResume(); pauser.onNext(true); } @Override protected void onPause() { pauser.onNext(false); super.onPause(); } } 

 Observable.interval(100, TimeUnit.MILLISECONDS).take(10) .buffer(Observable.interval(250, TimeUnit.MILLISECONDS)) .subscribe(System.out::println); 

 public class Event { public String jsonData; public String getJsonData() { return jsonData; } public Event setJsonData(String jsonData) { this.jsonData = jsonData; return this; } } 

 private PublishSubject<Event> eventQueue = PublishSubject.create(); 

  eventObservable = eventQueue.asObservable() .buffer(10, TimeUnit.SECONDS) //flush events every 10 seconds .toList() .doOnNext(new Action1<List<Event>>() { @Override public void call(List<Event> events) { apiClient.pushEvents(events); //push your event } }) .onErrorResumeNext(new Func1<Throwable, Observable<List<Event>>>() { @Override public Observable<List<Event>> call(Throwable throwable) { return null; //make sure, that on error will be never called } }) .subscribeOn(Schedulers.io()) .observeOn(Schedulers.io()); 

 eventSubscription = eventObservable.subscribe()