RxJS序列等价于promise.then()？

Question

我以前很有发展前途，现在我要搬到RxJS了。RxJS的文档并没有提供一个关于如何从承诺链转移到观察者序列的非常清楚的示例。

例如，我通常编写包含多个步骤的承诺链，如

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

我应该如何用RxJS风格重写这个承诺链？

Answer 1

对于数据流(相当于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
  });

一个承诺可以转化为可以通过Rx.Observable.fromPromise观察到的承诺。

一些承诺运算符有一个直接翻译。例如，RSVP.all或jQuery.when可以由Rx.Observable.forkJoin代替。

请记住，您有一组操作符，它们允许异步转换数据，并执行无法或很难执行的任务。Rxjs用异步数据序列(序列，即超过一个异步值)显示它的所有功能。

对于错误管理，主题稍微复杂一点。

• 还有捕捉和最后运算符
• 在出现错误时，retryWhen还可以帮助重复序列。
• 您还可以使用onError函数处理订阅服务器本身中的错误。

要获得精确的语义，请更深入地查看您可以在web上找到的文档和示例，或者在这里问一些特定的问题。

这无疑是深入使用Rxjs：handling.html进行错误管理的良好起点。

Answer 2

一种更现代的选择：

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

还请注意，要使所有这些都能工作，您需要在某个地方将subscribe连接到这个管道Observable，但我假设它是在应用程序的其他部分中处理的。

Answer 3

使用RxJs 6更新2019年5月

同意上面提供的答案，希望用一些玩具数据添加一个具体的例子&简单的承诺(用setTimeout)使用RxJs v6来增加清晰度。

只需将传递的id (当前硬编码为1)更新为不存在的东西，以执行错误处理逻辑。重要的是，还要注意of与catchError消息的使用。

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

产出数据：

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 }

关键部分，相当于以下使用普通承诺控制流程：

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