将承诺转换为可观察的

Question

我试图围绕着可观察者.我喜欢observables解决开发和可读性问题的方式.在我读到的时候,好处是巨大的.

关于HTTP和集合的可观察量似乎是直截了当的.我怎样才能将这样的东西转换为可观察的模式.

这是来自我的服务组件,用于提供身份验证.我希望它能像Angular2中的其他HTTP服务一样工作 - 支持数据,错误和完成处理程序.

firebase.auth().createUserWithEmailAndPassword(email, password)
  .then(function(firebaseUser) {
    // do something to update your UI component
    // pass user object to UI component
  })
  .catch(function(error) {
    // Handle Errors here.
    var errorCode = error.code;
    var errorMessage = error.message;
    // ...
  });

这里的任何帮助将不胜感激.我唯一的替代解决方案是创建EventEmitters.但我想这是在服务部门做事的一种可怕方式

Answer 1

如果您使用的是RxJS 6.0.0:

import { from } from 'rxjs';
const observable = from(promise);

Answer 2

试试这个:

import 'rxjs/add/observable/fromPromise';
import { Observable } from "rxjs/Observable";

const subscription = Observable.fromPromise(
    firebase.auth().createUserWithEmailAndPassword(email, password)
);
subscription.subscribe(firebaseUser => /* Do anything with data received */,
                       error => /* Handle error here */);

您可以在此处找到fromPromise运算符的完整参考

Answer 3

将 Promise 转换为 Observable 的正确模式defer是使用andfrom运算符：

import { defer, from } from 'rxjs';
    
const observable$ = defer(() => from(myPromise()));

为什么我们需要defer运营商？

Promise 是急切的，这意味着当被调用时它们会立即触发。这与可观察量的工作方式相反。Observables 是惰性的，它们仅在.subscribe()被调用时才会被触发。这就是我们需要始终将其包装到运算符中的原因defer。操作from员不做这项工作，所以defer总是需要的。

Answer 4

从

使用from直接转换之前创建的承诺，以可观察到的。

import { from } from 'rxjs';

// getPromise() will only be called once
const observable$ = from(getPromise());

创建Observable时，Promise的主体正在执行或已解决。如果内部Promise已解决，则Observable的新订户将立即获得其值（getPromise()不会再次调用）。

推迟

使用defer具有无极工厂功能输入到无极的创建和转换推迟到可观察到的。

import { defer } from 'rxjs';

// getPromise() will be called every time someone subscribes to the observable$
const observable$ = defer(() => getPromise());

与之不同的from是，defer等待订户，然后才通过调用给定的Promise工厂函数来创建新的Promise。当您要创建一个Observable但又不想立即执行内部Promise时，这很有用。内部的Promise仅在有人订阅Observable时执行。每个订户还将获得自己的新Observable。

from与defer示例之间的区别： https : //stackblitz.com/edit/rxjs-6rb7vf

const getPromise = val => new Promise(resolve => {
  console.log('Promise created for', val);
  setTimeout(() => resolve(`Promise Resolved: ${val}`), 5000);
});

// the execution of getPromise('FROM') starts here, when you create the promise inside from
const fromPromise$ = from(getPromise('FROM'));
const deferPromise$ = defer(() => getPromise('DEFER'));

fromPromise$.subscribe(console.log);
// the execution of getPromise('DEFER') starts here, when you subscribe to deferPromise$
deferPromise$.subscribe(console.log);

其他选择

创造可观测量（如大多数RxJS功能merge，concat，forkJoin，combineLatest...）和多家运营商（例如switchMap，mergeMap，concatMap，catchError...），也可以直接接受的承诺。

如果仍然使用其中一个，则不from必先包装Promise，尽管您仍可能希望使用它defer来实现仅在订阅上创建Promise的效果。

具体来说，大多数功能使用ObservableInput或SubscribableOrPromise定义如下。注意PromiseLike。

type ObservableInput<T> = SubscribableOrPromise<T> | ArrayLike<T> | Iterable<T>;

type SubscribableOrPromise<T> = Subscribable<T> | Subscribable<never> | PromiseLike<T> | InteropObservable<T>;

查看RxJS文档或实现，以查看您使用的功能是否支持ObservableInput或SubscribableOrPromise。