将promise转换为可观测

Question

我试着把我的头绕在可观察的物体上。我喜欢Observable解决开发和可读性问题的方式。正如我读到的，好处是巨大的。

HTTP和集合上的可观察对象似乎是直截了当的。我怎样才能把这样的东西转换成可观察的模式呢。

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

js prettyprint-override">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

使用from将先前创建的promise直接转换为可观察的promise。

import { from } from 'rxjs';

// getPromise() is called once, the promise is passed to the Observable
const observable$ = from(getPromise());

$将是有效地向订阅服务器重放Promises值的热可观察对象。

它是一个热门的可观察对象，因为生产者（在本例中是promise）是在可观察对象之外创建的。多个订户将共享同一promise。如果内部promise已经解决，可观察对象的新订户将立即获得其价值。

使用具有Promise工厂函数的延迟作为输入，以延迟创建和转换Promise为可观察对象。

import { defer } from 'rxjs';

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

$将是一个冷的可观察的。

它是一个冷的可观察对象，因为生产者（promise）是在可观察对象内部创建的。每个订户将通过调用给定的Promise factory函数来创建一个新的Promise。

这允许您创建一个可观察的$，而无需立即创建并执行promise，也无需与多个订户共享此promise。observable$的每个订户有效地从（promiseFactory（））调用。订阅（订户）。因此，每个订户创建并将其自己的新promise转换为新的可观察对象，并将自己附加到这个新的可观察对象上。

大多数RxJS运算符结合（例如合并，ConcatMap，forkJoin，组合最新...）或转换可观察对象（例如切换地图，合并地图，concatMap，>catchError...）直接接受promise。如果您使用其中一个，您不必首先使用from来包装promise（但是要创建冷可观察的，您可能仍然需要使用延迟）。

// Execute two promises simultaneously
forkJoin(getPromise(1), getPromise(2)).pipe(
  switchMap(([v1, v2]) => v1.getPromise(v2)) // map to nested Promise
)

检查留档或实现，以查看您使用的运算符是否接受观测输入或SubcribableOrPromise。

type ObservableInput<T> = SubscribableOrPromise<T> | ArrayLike<T> | Iterable<T>;
// Note the PromiseLike ----------------------------------------------------v
type SubscribableOrPromise<T> = Subscribable<T> | Subscribable<never> | PromiseLike<T> | InteropObservable<T>;

示例中，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);

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

如果您使用的是RxJS 6.0.0：

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