RxJS-Sequenz entsprechend versprechen.then ()?

83

Früher habe ich viel mit Versprechen entwickelt und jetzt ziehe ich zu RxJS. Das Dokument von RxJS bietet kein sehr klares Beispiel für den Übergang von der Versprechenskette zur Beobachtersequenz.

Zum Beispiel schreibe ich normalerweise eine Versprechenskette mit mehreren Schritten, wie z

// a function that returns a promise
getPromise()
.then(function(result) {
   // do something
})
.then(function(result) {
   // do something
})
.then(function(result) {
   // do something
})
.catch(function(err) {
    // handle error
});

Wie soll ich diese Versprechenskette im RxJS-Stil umschreiben?

Haoliang Yu
quelle

Antworten:

79

Für den Datenfluss (entspricht then):

Rx.Observable.fromPromise(...)
  .flatMap(function(result) {
   // do something
  })
  .flatMap(function(result) {
   // do something
  })
  .subscribe(function onNext(result) {
    // end of chain
  }, function onError(error) {
    // process the error
  });

Ein Versprechen kann in ein beobachtbares mit umgewandelt werden Rx.Observable.fromPromise.

Einige Versprechen Betreiber haben eine direkte Übersetzung. Zum Beispiel RSVP.alloder jQuery.whenkann ersetzt werden durch Rx.Observable.forkJoin.

Denken Sie daran, dass Sie über eine Reihe von Operatoren verfügen, mit denen Daten asynchron transformiert und Aufgaben ausgeführt werden können, die Sie mit Versprechungen nicht oder nur schwer ausführen können. Rxjs offenbart alle seine Kräfte mit asynchronen Datensequenzen (Sequenz dh mehr als 1 asynchroner Wert).

Für das Fehlermanagement ist das Thema etwas komplexer.

  • Es gibt Fang und schließlich auch Betreiber
  • retryWhen kann auch helfen, eine Sequenz im Fehlerfall zu wiederholen
  • Mit der onErrorFunktion können Sie auch Fehler im Teilnehmer selbst beheben .

Für eine genaue Semantik werfen Sie einen genaueren Blick auf die Dokumentation und Beispiele, die Sie im Internet finden, oder stellen Sie hier spezifische Fragen.

Dies wäre definitiv ein guter Ausgangspunkt, um das Fehlermanagement mit Rxjs zu vertiefen: https://xgrommx.github.io/rx-book/content/getting_started_with_rxjs/creating_and_querying_observable_sequences/error_handling.html

user3743222
quelle
Ich sehe immer, dass die beobachtbare Sequenz mit subscribe () endet. Gibt es einen Grund, dies zu tun, da dies nur eine Funktion eines beobachtbaren Objekts ist? Ist es die Funktion, die Sequenz zu starten?
Haoliang Yu
genau so. Wenn keine Beobachter das Abonnement durchlaufen haben, gibt Ihr Observable keine Daten aus, sodass Sie keinen Datenfluss sehen.
user3743222
7
Ich empfehle Ihnen, sich das anzuschauen: gist.github.com/staltz/868e7e9bc2a7b8c1f754 . IT könnte schmackhafter sein als das offizielle Dokument.
user3743222
3
Promise.thenist eher .flatMapals .map.
Tamas Hegedus
1
Zu Ihrer Information, dies ist nicht genau gleichwertig, da in der PromiseVersion Fehler vom 3. thenvon der abgefangen würden catch. Hier sind sie nicht.
mik01aj
34

Eine modernere Alternative:

import {from as fromPromise} from 'rxjs';
import {catchError, flatMap} from 'rxjs/operators';

fromPromise(...).pipe(
   flatMap(result => {
       // do something
   }),
   flatMap(result => {
       // do something
   }),
   flatMap(result => {
       // do something
   }),
   catchError(error => {
       // handle error
   })
)

Beachten Sie auch, dass Sie, subscribedamit dies alles funktioniert, Observableirgendwo auf diese Leitung zugreifen müssen, aber ich gehe davon aus, dass dies in einem anderen Teil der Anwendung behandelt wird.

mik01aj
quelle
Ich bin sehr neu in RxJS, aber da es sich hier nur um einen ersten Stream eines Ereignisses handelt und mergeMap()daher eigentlich nichts zusammenzuführen ist , glaube ich, dass wir in diesem Fall genau das Gleiche mit erreichen können concatMap()oder switchMap(). Habe ich das richtig verstanden ...?
Dan King
8

Update Mai 2019 mit RxJs 6

Stimmen Sie den oben angegebenen Antworten zu und möchten Sie ein konkretes Beispiel mit einigen Spielzeugdaten und einfachen Versprechungen (mit setTimeout) hinzufügen, indem Sie RxJs v6 verwenden , um Klarheit zu schaffen.

Aktualisieren Sie einfach die übergebene ID (derzeit fest codiert als 1) auf etwas, das nicht vorhanden ist, um auch die Fehlerbehandlungslogik auszuführen. Beachten Sie auch die Verwendung von ofwith catchErrormessage.

import { from as fromPromise, of } from "rxjs";
import { catchError, flatMap, tap } from "rxjs/operators";

const posts = [
  { title: "I love JavaScript", author: "Wes Bos", id: 1 },
  { title: "CSS!", author: "Chris Coyier", id: 2 },
  { title: "Dev tools tricks", author: "Addy Osmani", id: 3 }
];

const authors = [
  { name: "Wes Bos", twitter: "@wesbos", bio: "Canadian Developer" },
  {
    name: "Chris Coyier",
    twitter: "@chriscoyier",
    bio: "CSS Tricks and CodePen"
  },
  { name: "Addy Osmani", twitter: "@addyosmani", bio: "Googler" }
];

function getPostById(id) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      const post = posts.find(post => post.id === id);
      if (post) {
        console.log("ok, post found!");
        resolve(post);
      } else {
        reject(Error("Post not found!"));
      }
    }, 200);
  });
}

function hydrateAuthor(post) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      const authorDetails = authors.find(person => person.name === post.author);
      if (authorDetails) {
        post.author = authorDetails;
        console.log("ok, post hydrated with author info");
        resolve(post);
      } else {
        reject(Error("Author not Found!"));
      }
    }, 200);
  });
}

function dehydratePostTitle(post) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      delete post.title;
      console.log("ok, applied transformation to remove title");
      resolve(post);
    }, 200);
  });
}

// ok, here is how it looks regarding this question..
let source$ = fromPromise(getPostById(1)).pipe(
  flatMap(post => {
    return hydrateAuthor(post);
  }),
  flatMap(post => {
    return dehydratePostTitle(post);
  }),
  catchError(error => of(`Caught error: ${error}`))
);

source$.subscribe(console.log);

Ausgabedaten:

ok, post found!
ok, post hydrated with author info
ok, applied transformation to remove title
{ author:
   { name: 'Wes Bos',
     twitter: '@wesbos',
     bio: 'Canadian Developer' },
  id: 1 }

Der Schlüsselteil entspricht dem folgenden unter Verwendung eines einfachen Versprechungskontrollflusses:

getPostById(1)
  .then(post => {
    return hydrateAuthor(post);
  })
  .then(post => {
    return dehydratePostTitle(post);
  })
  .then(author => {
    console.log(author);
  })
  .catch(err => {
    console.error(err);
  });
arcseldon
quelle
0

wenn getPromiseFunktion ist in einer Mitte eines Stromrohres sollte man einfach wickelt es in einer der Funktionen mergeMap, switchMapoder concatMap( in der Regel mergeMap):

stream$.pipe(
   mergeMap(data => getPromise(data)),
   filter(...),
   map(...)
 ).subscribe(...);

Wenn Sie Ihren Stream mit starten möchten, getPromise()wickeln Sie ihn in folgende fromFunktion ein:

import {from} from 'rxjs';

from(getPromise()).pipe(
   filter(...)
   map(...)
).subscribe(...);
Maksim Romanenko
quelle
0

Soweit ich gerade herausgefunden habe, konvertiert es ein Ergebnis in einer flatMap in ein Array, selbst wenn Sie einen String zurückgegeben haben.

Wenn Sie jedoch ein Observable zurückgeben, kann dieses Observable einen String zurückgeben.

Samantha Adrichem
quelle
0

Wenn ich es richtig verstanden habe, meinst du damit, die Werte zu konsumieren. In diesem Fall verwendest du sbuscribe, dh

const arrObservable = from([1,2,3,4,5,6,7,8]);
arrObservable.subscribe(number => console.log(num) );

Darüber hinaus können Sie das Observable mit toPromise () wie gezeigt in ein Versprechen verwandeln:

arrObservable.toPromise().then()
David Kabii
quelle
0

So habe ich es gemacht.

Vorher

  public fetchContacts(onCompleteFn: (response: gapi.client.Response<gapi.client.people.ListConnectionsResponse>) => void) {
    const request = gapi.client.people.people.connections.list({
      resourceName: 'people/me',
      pageSize: 100,
      personFields: 'phoneNumbers,organizations,emailAddresses,names'
    }).then(response => {
      onCompleteFn(response as gapi.client.Response<gapi.client.people.ListConnectionsResponse>);
    });
  }

// caller:

  this.gapi.fetchContacts((rsp: gapi.client.Response<gapi.client.people.ListConnectionsResponse>) => {
      // handle rsp;
  });

Nach (ly?)

public fetchContacts(): Observable<gapi.client.Response<gapi.client.people.ListConnectionsResponse>> {
    return from(
      new Promise((resolve, reject) => {
        gapi.client.people.people.connections.list({
          resourceName: 'people/me',
          pageSize: 100,
          personFields: 'phoneNumbers,organizations,emailAddresses,names'
        }).then(result => {
          resolve(result);
        });
      })
    ).pipe(map((result: gapi.client.Response<gapi.client.people.ListConnectionsResponse>) => {
      return result; //map is not really required if you not changing anything in the response. you can just return the from() and caller would subscribe to it.
    }));
  }

// caller

this.gapi.fetchContacts().subscribe(((rsp: gapi.client.Response<gapi.client.people.ListConnectionsResponse>) => {
  // handle rsp
}), (error) => {
  // handle error
});
Anand Rockzz
quelle
Nebeneffekt : Die Änderungserkennung begann auch nach der Umwandlung des Rückrufs in eine beobachtbare Funktion .
Anand Rockzz
0

RxJS-Sequenz entsprechend versprechen.then ()?

Zum Beispiel

function getdata1 (argument) {
        return this.http.get(url)
            .map((res: Response) => res.json());
    }

    function getdata2 (argument) {
        return this.http.get(url)
            .map((res: Response) => res.json());
    }

    getdata1.subscribe((data1: any) => {
        console.log("got data one. get data 2 now");
        getdata2.subscribe((data2: any) => {
            console.log("got data one and two here");
        });
    });
Yogesh Waghmare
quelle