How to reset a BehaviorSubject

Question

I have a BehaviorSubject that I would like to reset - by that I mean I want the latest value to not be available, just as if it was just created.

I don't seem to see an API to do this but I suppose there is another way to achieve the same result?

My desired behavior is that I need to emit events, and I'd like subscribers to get the latest event when they subscribe - if a particular manager is in a 'started' state. But when this manager is 'stopped' the latest event should not be available (just like if it was never started in the first place).

Spunky

With a BehaviorSubject there's a next value available when it's just created, that's the whole point of it (and why you have to supply that value when you create it). – Cropeared

@Cropeared But you don't have to supply it when you create it! reactivex.io/RxJava/2.x/javadoc/io/reactivex/subjects/… – Spunky

Weird, it doesn't have that in the RxJS version (github.com/Reactive-Extensions/RxJS/blob/master/doc/api/…) - if you don't want to supply an initial value, why wouldn't you use a ReplaySubject instead? – Cropeared

I have the impression that BehaviorSubject is the equivalent of a ReplaySubject of size 1. But that doesn't answer my question at all :) – Spunky

I think the answer is: you can't. But I was also initially confused by the premise of an empty behavior subject! – Cropeared

If you need the most recent item that has emitted by your Subject before you subscribed to it - you may use BehaviorSubject but cannot use ReplaySubject. And yes, it can be empty when created. The initial value is useful as a "default" value. – Latish

Answer 1

I assume you want to clear the BehaviorSubject (because otherwise don't call onComplete on it). That is not supported but you can achieve a similar effect by having a current value that is ignored by consumers:

public static final Object EMPTY = new Object();

BehaviorSubject<Object> subject = BehaviorSubject.createDefault(EMPTY);

Observable<YourType> obs = subject.filter(v -> v != EMPTY).cast(YourType.class);

obs.subscribe(System.out::println);

// send normal data
subject.onNext(1);
subject.onNext(2);

// clear the subject
subject.onNext(EMPTY);

// this should not print anything
obs.subscribe(System.out::println);

Shaum

I could not use this solution, but it helped solving my issue as well: if you use generic classes like List e.g., the cast is not possible... Using an empty default object of the concrete class and filter with v -> !v.equals(EMPTY) and removing the cast should help in this case though. – Weathers

Answer 2

Another method of switching the value of an observable on and off is to use switchMap() to flip between the actual observable and an empty one.

Let's assume you have a manager object, and it has a observable that shows its state. Then,

subjectObservable = manager.getStateObservable()
  .switchMap( state -> state == ON ? subject : Observable.never() );

will only emit values while the manager is in the ON state.

Bonucci

Interesting! Thanks. – Spunky

Can you add how the manager maintains the stateObservable? – Geraldgeralda

stateObservable is maintained by the manager using whatever state information it has. The original question had the state as a BehaviorSubject, but any observable source could be used. – Bonucci

Answer 3

Just use setTimeout like this:

setOtpoint(value) {

    this._setOption.next(value);

    // Clear BehaviorSubject after emit value

    setTimeout(() => {
      this._setOption.next(null);
    }, 100);

  }

Gardant

i tried above code its working but after login again _setOption is get blank value and value is received after refresh – Allimportant

Answer 4

I find out a better solution for some cases and is:

subject.skiplast(1)

it can work to clean the last position on stream that is being retained because of the BehaviorSubject "behavior"

Milson

This returns an Observable though not a subject – Orphaorphan

Answer 5

A problem with @akarnokd's answer is that the .cast prevents YourType from being an interface or a generic type such as List<String>.

Another option is to filter on a boolean field that you can switch on and off.

    private BehaviorSubject<PandoraApp> subject = BehaviorSubject.create();
    private boolean enabled = true;

    Observable<PandoraApp> observable = subject.filter(v -> enabled);

If methods are being called on different threads you can use AtomicBoolean for the filter flag.

Inotropic

Answer 6

Here is my lib for this:

implementation "com.github.kolyall:rxjava2-empty:1.0.36"

Example:

private val myBehaviorSubject = BehaviorSubjectOptional.createOptional<MyItem?>()

errorBehaviorSubject.toObservable()
.subscribe{ item-> Log.d("onNext1", "item = $item")}

var item:MyItem? = MyItem()
myBehaviorSubject.onNextOptional(item)

//For reset:
myBehaviorSubject.clear()
//OR
item = null
myBehaviorSubject.onNextOptional(item)


errorBehaviorSubject.toObservable()
.subscribe{ item-> Log.d("onNext2", "item = $item")}

Doone