Angular2合并可观察对象

Question

我在处理一些 observables 时遇到了麻烦。我好像拿不到两个 observables 很好地放在一起。他们自己工作就好了，但我 需要这两个值。

db.glass.subscribe( ( glass: GlassData[] ): void => {
  console.log( glass ); // This prints
});

db.cassette_designs.subscribe( ( cassettes: CassetteData[] ): void => {
  console.log( cassettes ): // This prints
});

对 observables 不太熟悉，我尝试的第一件事是 将一个嵌套在另一个内部，但内部的似乎没有做任何事情。

db.glass.subscribe( ( glass: GlassData[] ): void => {
  console.log( glass ); // This prints

  db.cassette_designs.subscribe( ( cassettes: CassetteData[] ): void => {
    console.log( cassettes ): // This doesn't
  });
});

这似乎有点傻，所以我在谷歌上搜索，看看是否有 结合 observables 的更好方法，事实证明有一些。我试过了 zip 和 forkJoin 因为它们看起来最像我想做的，但是 他们也不行。

Observable.zip( db.cassette_designs, db.glass, ( cassettes: CassetteData[], glass: GlassData[] ): void => {
  console.log( cassettes ); // doesn't print
  console.log( glass     ); // doesn't print
});

Observable.forkJoin( [ db.cassette_designs, db.glass ] ).subscribe( ( data: any ): void => {
  console.log( data ); // doesn't print
});

这可能很简单，因为我没有正确调用函数，但是 我原以为我会在某个时候收到某种警告或错误。 tsc 代码没有任何问题，我在开发人员中没有收到任何消息 Chrome 或 Firefox 上的控制台。

更新

我试过combineLatest，但它仍然没有在控制台中显示任何内容。我一定错过了什么，但我不确定是什么。他们单独工作。

Observable.combineLatest( db.cassette_designs, db.glass, ( cassettes: CassetteData[], glass: GlassData[] ): void => {
  console.log( cassettes ); // doesn't print
  console.log( glass     ); // deson't print
});

observables 是通过以下方式创建的：

...

public Listen( event: string ): Observable<Response>
{
  return new Observable<Response>( ( subscriber: Subscriber<Response> ): Subscription => {
    const listen_func = ( res: Response ): void => subscriber.next( res );

    this._socket.on( event, listen_func );

    return new Subscription( (): void =>
      this._socket.removeListener( event, listen_func ) );
  });
}

...

然后，为了真正获得 observable，我发送了一个监听相关事件的响应。例如

...

public cassette_designs: Observable<CassetteData[]>;

...

this.cassette_designs = _socket.Listen( "get_cassette_designs" )
    .map( ( res: Response ) => res.data.data );

Answer 1

通过实际订阅结果，我设法让combineLatest 工作 可观察到。

原来我是这样做的：

Observable.combineLatest( db.cassette_designs, db.glass, ( cassettes: CassetteData[], glass: GlassData[] ): void => {
  console.log( cassettes );
  console.log( glass     );
});

现在我正在这样做：

Observable.combineLatest( db.cassette_designs, db.glass ).subscribe( ( data: any[] ): void => {
  console.log( data );
  // cassettes - data[0]
  // glass     - data[1]
});

Answer 2

跟进您的发现：

1) Observable 是一种惰性执行数据类型，这意味着它在订阅之前不会执行管道中的任何内容。这也适用于组合运算符。 zip、forkJoin、combineLatest 和 withLatestFrom 都将递归订阅 Observables，您仅在他们自己订阅之后才传递它们。

因此：

var output = Observable.combinelatest(stream1, stream2, (x, y) => ({x, y}));

在您调用output.subscribe() 之前实际上不会做任何事情，此时subscribe 也会在stream1 和stream2 上被调用，您将获得Rx 的所有魔力。

2) 更小的一点，但是每当您开始使用自己的创建方法时，请先查看文档以查看它是否已经存在。有Arrays、Promises、Node 样式回调和甚至标准event patterns 的静态创建方法。

因此你的Listen 方法可以变成：

public Listen<R>(event: string): Observable<R> {
  return Observable.fromEvent(this._socket, event);
}

Answer 3

嘿，如果您想要一个发出 2 个源的组合数据的单个流（可观察），请查看 combineLatest 方法。

Answer 4

它们两者：Observable.zip 和 Observable.forkJoin 都必须工作。 （就个人而言，我更喜欢 'forkJoin' - 它返回一个顺序相同的数组，就像你推送 observables 一样，而且你不需要很多参数）

也许，如果您手动创建 observables (Observable.create...)，您只是忘记从 'create' 方法为两个提供的 Observers 调用 'completed'。

Answer 5

我用过 combineLatest。我需要三个 API 调用，但我只需要一个对象来组合三个响应。

我遵循了同一篇文章中提到的公式：

var output = Observable.combinelatest(stream1, stream2, (x, y) => ({x, y}));

作为最终代码：

getGobalStats(): Observable<any> {

    let obs1 = this._http.get(this._config.getGlobalStatsUrl(), this._config.getOptions())
      .map((res: Response) => {
        return res.json().content;
      })
      .catch((error: any) => { console.error(error); return error; });
    let obs2 = this._http.get(this._config.getGlobalStatsUrl() + '?type=1', this._config.getOptions())
      .map((res: Response) => {
        return res.json().content;
      })
      .catch((error: any) => { console.error(error); return error; });
    let obs3 = this._http.get(this._config.getGlobalStatsUrl() + '?type=3', this._config.getOptions())
      .map((res: Response) => {
        return res.json().content;
      })
      .catch((error: any) => { console.error(error); return error; });

      return Observable.combineLatest(obs1,obs2,obs3,(res1,res2,res3) => { return {all:res1,running: res2, cycling: res3}});
  }

Answer 6

        Observable.merge(
            Observable.fromEvent(canvasEl, 'mousedown'), Observable.fromEvent(canvasEl, 'touchstart'))
            .switchMap((e) => {
                return Observable.merge(
                    Observable.fromEvent(canvasEl, 'mousemove').takeUntil(Observable.fromEvent(canvasEl, 'mouseup')),
                    Observable.fromEvent(canvasEl, 'touchmove').takeUntil(Observable.fromEvent(canvasEl, 'touchend')),
                )
                    .pairwise()
            })
            .subscribe(...