Wie kann ich auf ein Rx Observable „warten“?

Ich möchte in der Lage sein, auf ein Observable zu warten, zB

const source = Rx.Observable.create(/* ... */)
//...
await source;

Ein naiver Versuch führt dazu, dass der await sofort aufgelöst wird und die Ausführung nicht blockiert wird

Bearbeiten: Der Pseudocode für meinen vollständigen beabsichtigten Anwendungsfall lautet:

if (condition) {
  await observable;
}
// a bunch of other code

Ich verstehe, dass ich den anderen Code in eine andere separate Funktion verschieben und an den Subscribe-Callback übergeben kann, aber ich hoffe, das vermeiden zu können.

Sie müssen ein Versprechen abgeben await. Wandeln Sie das nächste Ereignis des Observables in ein Versprechen um und warten Sie darauf.

if (condition) {
  await observable.first().toPromise();
}

Hinweis bearbeiten: Diese Antwort verwendete ursprünglich .take(1), wurde aber geändert, um .first() zu verwenden, wodurch das Problem vermieden wird, dass das Promise niemals behoben wird, wenn der Stream endet, bevor ein Wert durchkommt.

Ab RxJS v8, toPromise wird entfernt. Stattdessen kann das obige durch ersetzt werden await firstValueFrom(observable)

Verwenden Sie das Neue firstValueFrom() oder lastValueFrom() anstatt toPromise(), die, wie hier erwähnt, ab RxJS 7 veraltet ist und in RxJS 8 entfernt wird.

import { firstValueFrom} from 'rxjs';
import { lastValueFrom } from 'rxjs';

this.myProp = await firstValueFrom(myObservable$);
this.myProp = await lastValueFrom(myObservable$);

Dies ist in RxJS 7+ verfügbar

Sehen: https://indepth.dev/rxjs-heads-up-toromise-is-being-deprecated/

Wahrscheinlich muss es sein

await observable.first().toPromise();

Wie bereits in den Kommentaren erwähnt, gibt es einen wesentlichen Unterschied zwischen take(1) und first() Operatoren, wenn es ein leeres abgeschlossenes Observable gibt.

Observable.empty().first().toPromise() führt zur Ablehnung mit EmptyError das kann entsprechend gehandhabt werden, weil es wirklich keinen Wert gab.

Und Observable.empty().take(1).toPromise() führt zur Auflösung mit undefined Wert.

Du wirst brauchen await ein Versprechen, also werden Sie verwenden wollen toPromise(). Sehen Dies für weitere Details auf toPromise().

Bearbeiten:

.toPromise() ist jetzt in RxJS 7 veraltet (Quelle: https://rxjs.dev/deprecations/to-promise)

Neue Antwort:

Als Ersatz für die veraltete Methode toPromise() sollten Sie eine der beiden integrierten statischen Konvertierungsfunktionen firstValueFrom oder lastValueFrom verwenden.

Beispiel:

import { interval, lastValueFrom } from 'rxjs';
import { take } from 'rxjs/operators';
 
async function execute() {
  const source$ = interval(2000).pipe(take(10));
  const finalNumber = await lastValueFrom(source$);
  console.log(`The final number is ${finalNumber}`);
}
 
execute();
 
// Expected output:
// "The final number is 9"

Alte Antwort:

Wenn toPromise für Sie veraltet ist, können Sie verwenden .pipe(take(1)).toPromise aber wie man sieht Hier es ist nicht veraltet.

Also bitte einfach benutzen toPromise (RxJs 6) wie gesagt:

//return basic observable
const sample = val => Rx.Observable.of(val).delay(5000);
//convert basic observable to promise
const example = sample('First Example')
  .toPromise()
  //output: 'First Example'
  .then(result => {
    console.log('From Promise:', result);
  });

async/await-Beispiel:

//return basic observable
const sample = val => Rx.Observable.of(val).delay(5000);
//convert basic observable to promise
const example = await sample('First Example').toPromise()
// output: 'First Example'
console.log('From Promise:', result);

Weiterlesen Hier.

Verwenden toPromise() wird nicht empfohlen, da es ab RxJs 7 abgeschrieben wird. Sie können zwei neue Operatoren verwenden, die in RxJs 7 vorhanden sind lastValueFrom() und firstValueFrom(). Weitere Einzelheiten können gefunden werden Hier

const result = await lastValueFrom(myObservable$);

Implementierungen in der Beta-Version sind hier verfügbar:

• ersterWertVon
• letzterWertVon

Ich verwende RxJS V 6.4.0, daher sollte ich eine veraltete Version in V 7.xx verwenden toPromise(). Inspiriert von anderen Antworten, hier ist, was ich damit gemacht habe toPromise()

import { first, ... } from 'rxjs/operators';

...

if (condition) {
  await observable$.pipe(first()).toPromise();
}

...

Beachten Sie, wie ich verwendet habe last() in einem pipe(). Denn auf meinem observable.first() funktioniert nicht wie von macil erwähnt

Hoffe, das hilft anderen, die RxJS V 6.xx wie ich verwenden :).

Danke.