Como obter o valor atual de RxJS Subject ou Observable?

Eu tenho um serviço 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);
    });
  }
}

Tudo funciona muito bem. Mas eu tenho outro componente que não precisa se inscrever, ele só precisa obter o valor atual de isLoggedIn em um determinado momento. Como posso fazer isso?

A Subjectou Observablenão tem um valor atual. Quando um valor é emitido, ele é passado aos assinantes e aoObservable é feito com ele.

Se você deseja ter um valor atual, use o BehaviorSubjectque foi projetado exatamente para esse fim.BehaviorSubjectmantém o último valor emitido e o emite imediatamente para novos assinantes.

Também possui um método getValue()para obter o valor atual.

A única maneira que você deve estar recebendo valores "fora de" um observável / Assunto é com se inscrever!

Se você estiver usando, getValue()estará fazendo algo imperativo no paradigma declarativo. Está lá como uma escotilha de fuga, mas 99,9% do tempo você NÃO deve usar getValue(). Existem algumas coisas interessantes quegetValue() farão: Irá gerar um erro se o assunto tiver sido cancelado, impedirá que você obtenha um valor se o assunto estiver morto por estar errado, etc. Mas, novamente, existe como uma fuga eclodem em circunstâncias raras.

Existem várias maneiras de obter o valor mais recente de um Assunto ou Observável da maneira "Rx-y":

1. Usando BehaviorSubject: Mas realmente assinando . Quando você assina pela primeira vezBehaviorSubject vez, enviará de forma síncrona o valor anterior que recebeu ou foi inicializado.
2. Usando um ReplaySubject(N): Isso armazenará em cacheN valores e reproduzi-las aos novos assinantes.
3. A.withLatestFrom(B): Use esse operador para obter o valor mais recente de observável Bquando Aemitido. Fornecerá os dois valores em uma matriz [a, b].
4. A.combineLatest(B): Use este operador para obter os valores mais recentes do Ae Bcada vez que quer Aou Bemite. Fornecerá os dois valores em uma matriz.
5. shareReplay(): Faz um multicast observável por meio de um ReplaySubject , mas permite repetir o observável por erro. (Basicamente, fornece esse comportamento de armazenamento em cache promissor).
6. publishReplay(), publishBehavior(initialValue), multicast(subject: BehaviorSubject | ReplaySubject), Etc: Outros operadores que a alavancagem BehaviorSubjecte ReplaySubject. Diferentes sabores da mesma coisa, eles basicamente fazem multicast da fonte observável, canalizando todas as notificações através de um assunto. Você precisa ligar connect()para se inscrever na fonte com o assunto.

Eu tive uma situação semelhante em que assinantes atrasados ​​assinam o Assunto após a chegada do valor.

Eu encontrei ReplaySubject, que é semelhante ao BehaviorSubject, funciona como um encanto neste caso. E aqui está um link para uma melhor explicação: 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
// .................
// .................
// .................

Você pode conferir o artigo completo sobre como implementá-lo aqui. https://www.imkrish.com/how-to-get-current-value-of-observable-in-a-clean-way/

Encontrei o mesmo problema em componentes filhos, nos quais, inicialmente, ele teria que ter o valor atual do Assunto, depois me inscrevi no Assunto para ouvir as alterações. Eu apenas mantenho o valor atual no Serviço para que ele esteja disponível para acesso aos componentes, por exemplo:

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

Um componente que precisa do valor atual pode acessá-lo a partir do serviço, ou seja:

sessionStorage.isLoggedIn

Não tenho certeza se esta é a prática correta :)

Um semelhante à procura resposta foi downvoted. Mas acho que posso justificar o que estou sugerindo aqui para casos limitados.

Embora seja verdade que um observável não tenha um valor atual , muitas vezes ele terá um valor imediatamente disponível . Por exemplo, nos repositórios redux / flux / akita, você pode solicitar dados de um repositório central, com base em vários observáveis ​​e esse valor geralmente estará disponível imediatamente.

Se for esse o caso, quando você subscribeo valor voltará imediatamente.

Então, digamos que você tenha telefonado para um serviço e, ao concluir, deseja obter o valor mais recente de algo da sua loja, que potencialmente não seja emitido :

Você pode tentar fazer isso (e manter o máximo possível as coisas 'dentro dos tubos'):

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

                        // we have serviceCallResponse and customer 
                     });

O problema é que ele será bloqueado até que o observável secundário emita um valor, o que potencialmente nunca poderia ser.

Recentemente, me vi precisando avaliar um observável apenas se um valor estivesse disponível imediatamente e, mais importante, eu precisava ser capaz de detectar se não estava. Acabei fazendo isso:

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

Observe que, para todas as opções acima, estou usando subscribepara obter o valor (como @Ben discute). Não .valueestou usando uma propriedade, mesmo que eu tivesse um BehaviorSubject.

Embora possa parecer um exagero, essa é apenas outra solução "possível" para manter o tipo Observable e reduzir o padrão ...

Você sempre pode criar um getter de extensão para obter o valor atual de um Observável.

Para fazer isso, você precisaria estender a Observable<T>interface em um global.d.tsarquivo de declaração de digitação. Em seguida, implemente o getter de extensão em um observable.extension.tsarquivo e, finalmente, inclua os tipos e o arquivo de extensão no seu aplicativo.

Você pode consultar esta Resposta do StackOverflow para saber como incluir as extensões no seu aplicativo 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...
  });

Você pode armazenar o último valor emitido separadamente do Observável. Em seguida, leia-o quando necessário.

let lastValue: number;

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

A melhor maneira de fazer isso é usar Behaviur Subject, aqui está um exemplo:

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

Uma assinatura pode ser criada e após a destruição do primeiro item emitido. Pipe é uma função que usa um Observable como entrada e retorna outro Observable como saída, sem modificar o primeiro observável. Angular 8.1.0. Pacotes: "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);
      });

    }
  }