Jak uzyskać bieżącą wartość przedmiotu RxJS lub obserwowalnego?

Mam usługę Angular 2:

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);
    });
  }
}

Wszystko działa świetnie. Ale mam inny komponent, który nie musi się subskrybować, musi tylko uzyskać bieżącą wartość isLoggedIn w pewnym momencie. W jaki sposób mogę to zrobić?

A Subjectlub Observablenie ma bieżącej wartości. Gdy emitowana jest wartość, jest ona przekazywana subskrybentom i Observablejest z nią wykonywana.

Jeśli chcesz mieć aktualną wartość, użyj tego, BehaviorSubjectco zostało zaprojektowane właśnie do tego celu. BehaviorSubjectzachowuje ostatnią emitowaną wartość i natychmiast emituje ją do nowych subskrybentów.

Ma również metodę getValue()uzyskania bieżącej wartości.

Jedyny sposób, w jaki powinny być coraz wartości „z” obserwowanej / Temat jest z zapisz się!

Jeśli używasz getValue(), robisz coś imperatywnego w deklaratywnym paradygmacie. Występuje jako właz ewakuacyjny, ale 99,9% czasu NIE powinno się go używać getValue(). Jest kilka interesujących rzeczy, które getValue()mogą zrobić: spowoduje zgłoszenie błędu, jeśli temat został wypisany z subskrypcji, zapobiegnie uzyskaniu wartości, jeśli podmiot nie żyje, ponieważ został zignorowany itp. Ale znowu jest to ucieczka wykluć się w rzadkich okolicznościach.

Istnieje kilka sposobów uzyskania najnowszej wartości od podmiotu lub obserwacji w sposób „Rx-y”:

1. Używanie BehaviorSubject: Ale faktycznie subskrybowanie . Po pierwszej subskrypcji BehaviorSubjectsynchronicznie wyśle ​​poprzednią wartość, którą otrzymała lub została zainicjowana.
2. Za pomocą ReplaySubject(N): Spowoduje to buforowanie Nwartości i odtworzenie ich nowym subskrybentom.
3. A.withLatestFrom(B): Użyj tego operatora, aby uzyskać najnowszą wartość z możliwej do zaobserwowania, Bgdy można ją zaobserwować A. Da ci obie wartości w tablicy [a, b].
4. A.combineLatest(B): Za pomocą tego operatora, aby uzyskać najnowsze wartości od Ai Bza każdym razem albo Aczy Bemituje. Da ci obie wartości w tablicy.
5. shareReplay(): Dokonuje obserwowalnej multiemisji przez ReplaySubject, ale umożliwia ponowienie próby zaobserwowania po błędzie. (Zasadniczo daje to obiecujące zachowanie buforowania).
6. publishReplay(), publishBehavior(initialValue), multicast(subject: BehaviorSubject | ReplaySubject), Etc: Inni operatorzy że dźwignia BehaviorSubjecta ReplaySubject. Różne smaki tej samej rzeczy, w zasadzie multiemisję źródła można zaobserwować, kierując wszystkie powiadomienia za pośrednictwem tematu. Musisz zadzwonić, connect()aby zasubskrybować źródło z tematem.

Miałem podobną sytuację, w której spóźnieni subskrybenci subskrybują Temat po otrzymaniu jego wartości.

Znalazłem ReplaySubject, który jest podobny do BehaviorSubject działa w tym przypadku jak urok. A oto link do lepszego wyjaśnienia: http://reactivex.io/rxjs/manual/overview.html#replaysubject

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
// .................
// .................
// .................

Możesz sprawdzić pełny artykuł o tym, jak go wdrożyć tutaj. https://www.imkrish.com/how-to-get-current-value-of-observable-in-a-clean-way/

Ten sam problem napotkałem w komponentach potomnych, gdzie początkowo musiałby mieć bieżącą wartość Przedmiotu, a następnie zasubskrybować Temat, aby słuchać zmian. Po prostu utrzymuję bieżącą wartość w usłudze, aby była dostępna dla komponentów, na przykład:

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;
  }
}

Komponent, który potrzebuje bieżącej wartości, może właśnie uzyskać dostęp do niej z usługi, tj .:

sessionStorage.isLoggedIn

Nie jestem pewien, czy to dobra praktyka :)

Podobna patrząc odpowiedź została downvoted. Ale myślę, że mogę uzasadnić to, co sugeruję tutaj w ograniczonych przypadkach.

Chociaż prawdą jest, że obserwowalne nie ma bieżącej wartości, bardzo często będzie miało natychmiast dostępną wartość. Na przykład w sklepach redux / flux / akita możesz zażądać danych ze sklepu centralnego, w oparciu o pewną liczbę obserwowalnych, a ta wartość będzie na ogół natychmiast dostępna.

Jeśli tak jest, to kiedy Ty subscribe, wartość natychmiast wróci.

Powiedzmy, że masz połączenie z usługą, a po zakończeniu chcesz uzyskać najnowszą wartość czegoś ze swojego sklepu, który potencjalnie może nie emitować :

Możesz spróbować to zrobić (i powinieneś w miarę możliwości trzymać rzeczy „wewnątrz rur”):

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

                        // we have serviceCallResponse and customer 
                     });

Problem polega na tym, że blokuje się, dopóki obserwowalne wtórne nie wyemituje wartości, która potencjalnie może nigdy nie być.

Niedawno stwierdziłam, że muszę ocenić obserwowalne tylko wtedy, gdy wartość jest natychmiast dostępna , a co ważniejsze, muszę być w stanie wykryć, czy nie była. Skończyło się na tym, że:

 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');
                        }
                     });

Zauważ, że dla wszystkich powyższych używam, subscribeaby uzyskać wartość (jak omawia @Ben). Nie korzystam z .valuenieruchomości, nawet gdybym miał BehaviorSubject.

Chociaż może to zabrzmieć przesadnie, jest to kolejne „możliwe” rozwiązanie pozwalające zachować typ obserwowalny i zmniejszyć płytę kotłową ...

Zawsze możesz utworzyć moduł pobierający rozszerzenia, aby uzyskać bieżącą wartość Obserwowalnego.

Aby to zrobić, musisz rozszerzyć Observable<T>interfejs w global.d.tspliku deklaracji typów. Następnie wdrożyć getter rozszerzenie w observable.extension.tspliku i wreszcie obejmować zarówno typowania i plik rozszerzenia do aplikacji.

Możesz zapoznać się z tą odpowiedzią StackOverflow, aby dowiedzieć się, jak dołączyć rozszerzenia do aplikacji Angular.

// global.d.ts
declare module 'rxjs' {
  interface Observable<T> {
    /**
     * _Extension Method_ - Returns current value of an Observable.
     * Value is retrieved using _first()_ operator to avoid the need to unsubscribe.
     */
    value: Observable<T>;
  }
}

// observable.extension.ts
Object.defineProperty(Observable.prototype, 'value', {
  get <T>(this: Observable<T>): Observable<T> {
    return this.pipe(
      filter(value => value !== null && value !== undefined),
      first());
  },
});

// using the extension getter example
this.myObservable$.value
  .subscribe(value => {
    // whatever code you need...
  });

Możesz przechowywać ostatnią emitowaną wartość oddzielnie od Obserable. Następnie przeczytaj go w razie potrzeby.

let lastValue: number;

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

Najlepszym sposobem na to jest użycie Behaviur Subject, oto przykład:

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

Subskrypcję można utworzyć i po zniszczeniu pierwszego emitowanego przedmiotu. Rura jest funkcją, która wykorzystuje Obserowalny jako swój sygnał wejściowy i zwraca inny Obserowalny jako wynik, nie modyfikując jednak pierwszego obserwowalnego. Kątowy 8.1.0. Pakiety: "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);
      });

    }
  }