How to get current value of RxJS Subject or Observable?

Question

I have an Angular 2 service:

import {Storage} from './storage';
import {Injectable} from 'angular2/core';
import {Subject}    from 'rxjs/Subject';

@Injectable()
export class SessionStorage extends Storage {
  private _isLoggedInSource = new Subject<boolean>();
  isLoggedIn = this._isLoggedInSource.asObservable();
  constructor() {
    super('session');
  }
  setIsLoggedIn(value: boolean) {
    this.setItem('_isLoggedIn', value, () => {
      this._isLoggedInSource.next(value);
    });
  }
}

Everything works great. But I have another component which doesn't need to subscribe, it just needs to get the current value of isLoggedIn at a certain point in time. How can I do this?

Hebner

Answer 1

A Subject or Observable doesn't have a current value. When a value is emitted, it is passed to subscribers and the Observable is done with it.

If you want to have a current value, use BehaviorSubject which is designed for exactly that purpose. BehaviorSubject keeps the last emitted value and emits it immediately to new subscribers.

It also has a method getValue() to get the current value.

Lumber

what is the difference than simply doing $behaviorSubject.value rather than $behaviorSubject.getValue() ? From the source code, the getValue() method does some checking. github.com/Reactive-Extensions/RxJS/blob/master/src/core/… – Fiedler

To me it looks like the main difference is that with getValue() an exception will be thrown when the last value was an error, while value will onky give the last non-error value. – Harold

Hi, it is my bad, they are the same thing now in the rxjs 5. The source code link above was referring to rxjs4. – Fiedler

Important note from the author of RxJS 5: Using getValue() is a HUGE red flag you're doing something wrong. It's there as an escape hatch. Generally everything you do with RxJS should be declarative. getValue() is imperative. If you're using getValue(), there's a 99.9% chance you're doing something wrong or weird. – Joby

@BenLesh what if I have a BehaviorSubject<boolean>(false) and like to toggle it? – Archil

@Archil just emit a new value – Harold

@GünterZöchbauer What I meant was: when emitting a new boolean, I need to know the current value to toggle it. In that case, is using getValue() justified? – Archil

you can also store the value in a field, but I think using getValue is fine – Harold

You don't need getValue to toggle it. Just use foo$.pipe(take(1)).subscribe(val => foo$.next(!val)). Since it's a behavior subject, this will work instantly as you get the last emitted value on subscription. – Frag

@BenLesh getValue() is very useful for say, doing an instantaneous action, like onClick->dispatchToServer(x.getValue()) but don't use it in observable chains. – Manaus

@IngoBürk The only way I can justify your suggestion is if you didn't know that foo$ was a BehaviorSubject (i.e. it is defined as Observable and maybe it's hot or cold of from a deferred source). However since you then go on to use .next on $foo itself that means your implementation relies on it being a BehaviorSubject so there's no justification for not just using .value to get the current value in the first place. – Cusack

@Cusack it relies on it being a Subject, not a BehaviorSubject. getValue, on the other hand, does tie it to the specific subject implementation. I've seen it over and over again that Behavior* is used where Replay* would be appropriate, and this makes refactoring that very difficult. – Frag

Thanks for this, RxJS is pretty hard to understand, but once you get it, you cannot do without it haha – Subatomic

@GünterZöchbauer taking into account the input from Ben Lesh would you consider rephrasing your answer? It encourages to use BehaviourSubject as a pure value container rather than Observable. – Vidar

@Vidar I don't think it does encourage anything. I just explained a way to get what is asked in the the question. I don't like answers that say "don't do that" if there actually is a way. What the answer from Ben Lesh states is correct, but doesn't answer the question. – Harold

Answer 2

The only way you should be getting values "out of" an Observable/Subject is with subscribe!

If you're using getValue() you're doing something imperative in declarative paradigm. It's there as an escape hatch, but 99.9% of the time you should NOT use getValue(). There are a few interesting things that getValue() will do: It will throw an error if the subject has been unsubscribed, it will prevent you from getting a value if the subject is dead because it's errored, etc. But, again, it's there as an escape hatch for rare circumstances.

There are several ways of getting the latest value from a Subject or Observable in a "Rx-y" way:

1. Using BehaviorSubject: But actually subscribing to it. When you first subscribe to BehaviorSubject it will synchronously send the previous value it received or was initialized with.
2. Using a ReplaySubject(N): This will cache N values and replay them to new subscribers.
3. A.withLatestFrom(B): Use this operator to get the most recent value from observable B when observable A emits. Will give you both values in an array [a, b].
4. A.combineLatest(B): Use this operator to get the most recent values from A and B every time either A or B emits. Will give you both values in an array.
5. shareReplay(): Makes an Observable multicast through a ReplaySubject, but allows you to retry the observable on error. (Basically it gives you that promise-y caching behavior).
6. publishReplay(), publishBehavior(initialValue), multicast(subject: BehaviorSubject | ReplaySubject), etc: Other operators that leverage BehaviorSubject and ReplaySubject. Different flavors of the same thing, they basically multicast the source observable by funneling all notifications through a subject. You need to call connect() to subscribe to the source with the subject.

Joby

can you elaborte why using getValue() is a red flag? What if i need the current value of the observable only once, the moment the user clicks a button? Should I subscribe for the value and immediately unsubscribe? – Waiter

It's okay sometimes. But often it's a sign that the code's author is doing something very imperative (usually with side-effects) where they shouldn't be. You can do click$.mergeMap(() => behaviorSubject.take(1)) to solve your problem as well. – Joby

Great explanation! Actually, if you can keep Observable and use the methods, it's a much better way. In a context of typescript with Observable used everywhere but only a few one defined as BehaviourSubject, then it would be less consistant code. The methods you proposed allowed me to keep Observable types everywhere. – Cupboard

its good if you want to simply dispatch the current value (like send it to the server on click), doing a getValue() is very handy. but don't use it when chaining observable operators. – Manaus

It's fine if you're just firing an action for ngrx to decide what to do with later. – Manaus

I have an AuthenticationService which uses a BehaviourSubject to store the current logged in state (boolean true or false). It exposes an isLoggedIn$ observable for subscribers who want to know when the state changes. It also exposes a get isLoggedIn() property, which returns the current logged in state by calling getValue() on the underlying BehaviourSubject - this is used by my authentication guard to check the current state. This seems like a sensible use of getValue() to me...? – Mawkish

@DanKing from your route guard you can return an observable like so: return isLoggedIn$.pipe(map(loggedIn => loggedIn ? true : this.router.parseUrl('/'))); – Whitehot

@BenLesh doing the click$.mergeMap(... requires you to have the click$ as an observable. For it you have to put a template variable onto the element, get him into the class via ViewChild, convert it's nativeElement into an Observable using fromEvent... See what it looks like? Right, getElementById and addEventListener :D In a modern UI framework, omg... IMHO using BehaviorSubject.value is the lesser evil – Lindeman

@BenLesh can you link to where in the official docs it says what you are saying in point 1. that the previous value of BehaviourSubject is sent synchronously. – Aynat

Answer 3

I had similar situation where late subscribers subscribe to the Subject after its value arrived.

I found ReplaySubject which is similar to BehaviorSubject works like a charm in this case. And here is a link to better explanation: http://reactivex.io/rxjs/manual/overview.html#replaysubject

Basic

that helped me in my Angular4 app - I had to move the subscription from the component constructor to the ngOnInit() (such component is shared across routes), just leaving here in case someone has a similar issue – Kingkingbird

I had an issue with an Angular 5 app where I was using a service to get values from an api and set variables in different components. I was using subjects/observables but it wouldn't push the values after a route change. ReplaySubject was a drop in replacement for Subject and solved everything. – Cribbing

Make sure you're using ReplaySubject(1) otherwise new subscribers will get every previously emitted value in sequence - this isn't always obvious at runtime – Overeager

@Overeager as far as I understand ReplaySubject(1) behave the same as BehaviorSubject() – Basic

Not quite the same, the big difference being that ReplaySubject does not immediately emit a default value when it's subscribed to if it's next() function hasn't yet been invoked, whereas BehaviourSubject does. The immediate emit is very handy for applying default values to a view, when the BehaviourSubject is used as the data source observable in a service for example – Overeager

Answer 4

const observable = of('response')

function hasValue(value: any) {
  return value !== null && value !== undefined;
}

function getValue<T>(observable: Observable<T>): Promise<T> {
  return observable
    .pipe(
      filter(hasValue),
      first()
    )
    .toPromise();
}

const result = await getValue(observable)
// Do the logic with the result
// .................
// .................
// .................

You can check the full article on how to implement it from here. https://www.imkrish.com/blog/development/simple-way-get-value-from-observable

Primogenial

Exactly what I was looking for. Thank you! – Forecastle

Unfortunately link expired. I were used getValue() and this solution resolved my problems. Thanks! – Rucker

See firstValueFrom. Saves you from having to write the function yourself – Stronghold

Answer 5

I encountered the same problem in child components where initially it would have to have the current value of the Subject, then subscribe to the Subject to listen to changes. I just maintain the current value in the Service so it is available for components to access, e.g. :

import {Storage} from './storage';
import {Injectable} from 'angular2/core';
import {Subject}    from 'rxjs/Subject';

@Injectable()
export class SessionStorage extends Storage {

  isLoggedIn: boolean;

  private _isLoggedInSource = new Subject<boolean>();
  isLoggedIn = this._isLoggedInSource.asObservable();
  constructor() {
    super('session');
    this.currIsLoggedIn = false;
  }
  setIsLoggedIn(value: boolean) {
    this.setItem('_isLoggedIn', value, () => {
      this._isLoggedInSource.next(value);
    });
    this.isLoggedIn = value;
  }
}

A component that needs the current value could just then access it from the service, i.e,:

sessionStorage.isLoggedIn

Not sure if this is the right practice :)

Neale

If you need the value of an observable in your component view, you can just use the async pipe. – Frag

Answer 6

A similar looking answer was downvoted. But I think I can justify what I'm suggesting here for limited cases.

---

While it's true that an observable doesn't have a current value, very often it will have an immediately available value. For example with redux / flux / akita stores you may request data from a central store, based on a number of observables and that value will generally be immediately available.

If this is the case then when you subscribe, the value will come back immediately.

So let's say you had a call to a service, and on completion you want to get the latest value of something from your store, that potentially might not emit:

You might try to do this (and you should as much as possible keep things 'inside pipes'):

 serviceCallResponse$.pipe(withLatestFrom(store$.select(x => x.customer)))
                     .subscribe(([ serviceCallResponse, customer] => {

                        // we have serviceCallResponse and customer 
                     });

The problem with this is that it will block until the secondary observable emits a value, which potentially could be never.

I found myself recently needing to evaluate an observable only if a value was immediately available, and more importantly I needed to be able to detect if it wasn't. I ended up doing this:

 serviceCallResponse$.pipe()
                     .subscribe(serviceCallResponse => {

                        // immediately try to subscribe to get the 'available' value
                        // note: immediately unsubscribe afterward to 'cancel' if needed
                        let customer = undefined;

                        // whatever the secondary observable is
                        const secondary$ = store$.select(x => x.customer);

                        // subscribe to it, and assign to closure scope
                        sub = secondary$.pipe(take(1)).subscribe(_customer => customer = _customer);
                        sub.unsubscribe();

                        // if there's a delay or customer isn't available the value won't have been set before we get here
                        if (customer === undefined) 
                        {
                           // handle, or ignore as needed
                           return throwError('Customer was not immediately available');
                        }
                     });

Note that for all of the above I'm using subscribe to get the value (as @Ben discusses). Not using a .value property, even if I had a BehaviorSubject.

Cusack

Btw this works because by default ‘schedule’ uses the current thread. – Cusack

Answer 7

A subscription can be created, then after taking the first emitted item, destroyed. In the example below, pipe() is a function that uses an Observable as its input and returns another Observable as its output, while not modifying the first observable.

Sample created with Angular 8.1.0 packages "rxjs": "6.5.3", "rxjs-observable": "0.0.7"

  ngOnInit() {

    ...

    // If loading with previously saved value
    if (this.controlValue) {

      // Take says once you have 1, then close the subscription
      this.selectList.pipe(take(1)).subscribe(x => {
        let opt = x.find(y => y.value === this.controlValue);
        this.updateValue(opt);
      });

    }
  }

Algology

Answer 8

There are two ways you can achieve this.

1. BehaviorSubject has a method getValue() which you can get the value in a specific point of time.

2. You can subscribe directly with the BehaviorSubject and you may pass the subscribed value to a class member, field or property.

I wouldn't recommend both approaches.

In the first approach, it's a convenient method you can get the value anytime, you may refer to this as the current snapshot at that point of time. Problem with this is you can introduce race conditions in your code, you may invoke this method in many different places and in different timing which is hard to debug.

The second approach is what most developers employ when they want a raw value upon subscription, you can track the subscription and when do you exactly unsubscribe to avoid further memory leak, you may use this if you're really desperate to bind it to a variable and there's no other ways to interface it.

I would recommend, looking again at your use cases, where do you use it? For example you want to determine if the user is logged in or not when you call any API, you can combine it other observables:

const data$ = apiRequestCall$().pipe(
 // Latest snapshot from BehaviorSubject.
 withLatestFrom(isLoggedIn),
 // Allow call only if logged in.
 filter(([request, loggedIn]) => loggedIn)
 // Do something else..
);

With this, you may use it directly to the UI by piping data$ | async in case of angular.

Dormant

Answer 9

Another approach, If you want / can to use async await (has to be inside of an async functions) you can do this with modern Rxjs:

 async myFunction () {
     const currentValue = await firstValueFrom(
      of(0).pipe(
        withLatestFrom(this.yourObservable$),
        map((tuple) => tuple[1]),
        take(1)
      )
    );
    // do stuff with current value

 }

 

This will emit a value "Right away" because of withLatestFrom, and then will resolve the promise.

Urethroscope

Answer 10

You could store the last emitted value separately from the Observable. Then read it when needed.

let lastValue: number;

const subscription = new Service().start();
subscription
    .subscribe((data) => {
        lastValue = data;
    }
);

Convalescence

It's not a reactive approach to store some stuff outside the observable. Instead you should have as much data as possible flowing inside the observable streams. – Shondrashone

Answer 11

The best way to do this is using Behavior Subject, here is an example:

var sub = new rxjs.BehaviorSubject([0, 1])
sub.next([2, 3])
setTimeout(() => {sub.next([4, 5])}, 1500)
sub.subscribe(a => console.log(a)) //2, 3 (current value) -> wait 2 sec -> 4, 5

Cinematography