来自 EventEmitter 的热门和共享 Observable

Question

有没有办法从 EventEmitter （或 Angular 2 alpha 46 / RxJS 5 alpha 中提供的等效项）获得一个热可观察对象？即如果我们在解析值后订阅，它会触发之前解析的值。类似于我们总是返回相同的承诺。

理想情况下，仅使用 Angular 2 对象（我在某处读到稍后会嵌入轻量级 RxJS 以消除依赖关系），否则导入 RxJS 就可以了。 AsyncSubject 似乎符合我的需要，但它在 RxJS 5 alpha 中不可用。

我尝试了以下操作，但没有成功（从不触发）。知道如何使用它吗？

let emitter = new EventEmitter<MyObj>();
setTimeout(() => {emitter.next(new MyObj());});
this.observable = emitter;
return this.observable.share();

Full plunker here comparing hot and cold

用例：仅访问某些异步对象一次（例如，一系列 HTTP 调用合并/包装在 new EventEmitter 中），但提供已解析的异步对象订阅它的任何服务/组件，即使它们在解决后订阅（收到 HTTP 响应）。

编辑：问题不在于如何合并 HTTP 响应，而是如何从 EventEmitter 或任何可用于 Angular 2 alpha 46 / RxJS 5 alpha 的等价物获得（热？）可观察的，允许在检索/解析异步结果（HTTP 只是异步源的一个示例）。 myEventEmitter.share() 不起作用（参见上面的 plunker），尽管它适用于 HTTP 返回的 Observable（参见 plunker from@Eric Martinez）。从 Angular 2 alpha 46 开始，.toRx() 方法不再存在，EventEmitter 是可观察对象和主体本身。

只要我们总是返回相同的 Promise 对象，这与 Promise 配合得很好。由于我们在 HTTP Angular 2 服务中引入了观察者，因此我想避免混淆承诺和观察者（据说观察者比承诺更强大，所以它应该允许用承诺做容易的事情）。

Specs about share()（我还没有找到版本 5 alpha 的文档 - Angular 2 使用的版本） - 处理 Angular 2 HTTP 服务返回的 Observable，而不是 EventEmitter。

编辑：澄清了为什么不使用 HTTP 返回的 Observable，并补充说不直接使用 RxJS 会更好。

编辑：更改描述：关注的是多个订阅，而不是合并 HTTP 结果。

谢谢！

Answer 1

您似乎描述的功能不是冷可观察的功能，而是Rx.BehaviourSubject 的功能。在这里查看关于 Rxjs 主题的解释：https://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/subjects.md。

我从那里引用：

BehaviourSubject 类似于 ReplaySubject，只是它只存储它发布的最后一个值。 BehaviourSubject 在初始化时也需要一个默认值。当主体还没有收到其他值时，这个值被发送给观察者。这意味着所有订阅者将在订阅时立即收到一个值，除非主题已经完成。

Rx.AsyncSubject 将是最接近承诺的行为：

AsyncSubject 类似于 Replay 和 Behavior 主题，但是它只会存储最后一个值，并且只有在序列完成时才发布它。当源 observable 很热并且可能在任何观察者订阅它之前完成时，您可以使用 AsyncSubject 类型。在这种情况下，AsyncSubject 仍然可以提供最后一个值并将其发布给任何未来的订阅者。

另外两个cmets：

• 在你的插件中：this._coldObservable = emitter.share();。使用 share 返回一个 hot observable！
• EventEmitter 实际上首先扩展了主题

更新： 将 EventEmitter 包裹在 Rx.Observable 周围：

function toRx ( eventEmitter ) {
  return Rx.Observable.create(function ( observer ) {
    eventEmitter.subscribe(function listener ( value ) {observer.onNext(value)});
    // Ideally you also manage error and completion, if that makes sense with Angular2
    return function () {
      /* manage end of subscription here */
    };
  };
)
}

一旦你拥有Rx.Observable，你就可以申请share()、shareReplay(1)，任何你想要的。

我敢打赌，Angular 团队迟早会提出一个桥接功能，但如果你不想等待，你可以自己做。

Answer 2

ReplaySubject 正在做我正在寻找的事情。 @robwormald 提供了一个关于 gitter 的工作示例，我稍作修改以便更好地演示。

暴露 HTTP 响应：

import {Injectable} from 'angular2/angular2';
import {Http} from 'angular2/http';
import {ReplaySubject} from '@reactivex/rxjs/dist/cjs/Rx'

@Injectable()
export class PeopleService {
  constructor(http:Http) {
    this.people = new ReplaySubject(1);

    http.get('api/people.json')
      .map(res => res.json())
      .subscribe(this.people);
  }
}

多次订阅：

// ... annotations
export class App {
  constructor(peopleService:PeopleService) {

    people.subscribe(v => {
      console.log(v)
    });

    //some time later

    setTimeout(() => {
      people.subscribe(v => {
        console.log(v)
      });
      people.subscribe(v => {
        console.log(v)
      });
    },2000)
  }
}

Full plunker

编辑：BehaviorSubject 是另一种选择。在这个用例中，区别在于初始值，例如，如果我们想在使用 HTTP 响应更新之前显示缓存中的内容。