如何从 XHR 创建 Observable

Question

我正在尝试在 Angular2 中进行文件上传。我使用在这里找到的代码 - https://plnkr.co/edit/ozZqbxIorjQW15BrDFrg?p=info。

我需要在我的史诗（Angular2 + Redux）中使用响应，所以我想做类似的事情：

updateAvatarRequest = (action$: ActionsObservable<IPayloadAction>) => {
    return action$.ofType(MeActions.UPDATE_AVATAR)
      .mergeMap<IPayloadAction>(({payload}) => {
        return this.meService.updateAvatar()
          .map(result => ({
            // ACTION ON SUCCESS
          }))
          .catch(error => Observable.of({
            // ACTION ON ERROR
          }));
        });
}

问题是我没有从我的 Observable 得到任何响应，直到我不订阅它。另一个问题是我不能在这个 observable 上使用 catch()。如果我这样做，它会起作用：

this.meService.updateUserAvatar(formData)
    .subscribe(
      (x) => console.log('onNext: %s', x),
      (e) => console.log('onError: %s', e),
      () => console.log('onCompleted')
);

但我想保持一致。请帮助我处理这个 RxJS，直到我失去其余的头发。

Answer 1

在我不订阅之前，我的 Observable 没有收到任何响应

这是设计使然。我想action$.ofType(MeActions.UPDATE_AVATAR) 返回一个冷的 Observable，所以在至少有一个订阅者之前它不会被触发。您可以使用 publish() 将其转换为 ConnectableObservable 并立即调用 connect() 这使其成为热门的 Observable (https://jsbin.com/faquhew/5/edit?js,console)，但我认为最好坚持使用默认功能。

更多信息见：

• RxJS - .subscribe() vs .publish().connect()

• Hot and Cold observables : are there 'hot' and 'cold' operators?

• https://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/creating.md#cold-vs-hot-observables

Operator catch() 让我们继续使用从其回调返回的 Observable 链，请参阅：https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/catch.md

所以这取决于你想用catch() 实现什么。

编辑：

var source = Observable.create(obs => {
    obs.next(1);
    obs.next(2);
    obs.error(new Error('original error message'));
  })
  .catch(error => Observable.of('recovered from error'));

source.subscribe(val => console.log(val));

这会打印到控制台：

1
2
"recovered from error"

观看现场演示：https://jsbin.com/micebeg/4/edit?js,console