RxJava: A scan with Single based accumulator which disposes intermediate Single

user_258848 0
Viewed 506

I'm writing an Rx-based unidirectional UI flow where each state reduce is a Single. Usually such flows are done with scan (they need previous state), but when Single is involved it's a bit tricky. I managed to get it working like this:

val events = Observable.just("event1", "event2", "event3")
val initialState = Single.just(emptyList<String>())
// given a current state produces next state's Single
val reducer = { currentState: List<String>, event: String ->
  Single.fromCallable { /* do work */ currentState.plus(event) }
}

events
  .scan(
    initialState,
    { currentStateSingle, event ->
      val nextStateSingle = currentStateSingle
        .flatMap { curState -> reducer(curState, event) }
      // cache is required to avoid resubscription 
      // to all previously emitted single's on each new scan iteration
      nextStateSingle.cache()
    }
  )
  .flatMapSingle { it }
  .subscribe { state -> println("state updated to $state") }

What bothers me is that each event (of which there can be many in UI environment) will create a nextStateSingle.cache() and forever add it to an existing chain and all those Singles ever emitted will stay there, unboundedly consuming memory and never being disposed while after they emitted a new state once, they are not needed at all.

I've been thinking on how to do this with some kind of switchMap usage, or even using some external atomic variable to hold state (instead of scan), but I fail to find a way.

The only other option I see is to write a custom operator which will subscribe to inner single, wait for result then dispose it, but I'd like to avoid writing a custom operator.

rx-java2 rx-java reactive-programming android java
1 Answers

According to the documentation of cache, since you can't dispose the origin and you can't clear the cached values you can use this workaround which can control the cache and clear chached values by forgetting all references:

AtomicBoolean shouldStop = new AtomicBoolean();

source.takeUntil(v -> shouldStop.get())
        .onTerminateDetach()
        .cache()
        .takeUntil(v -> shouldStop.get())
        .onTerminateDetach()
        .subscribe(...);

Then maybe you can save the state and nulls out references once in a while.

user_64645130
Related