Angular2 Merging observables

Question

I'm having trouble getting with some observables. I can't seem to get two observables to place nice together. They work just fine on their own, but I need both values.

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

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

Not being all that familiar with observables, the first thing I tried was nesting one inside the other, but the inner one doesn't seem to do anything.

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

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

This seemed a bit silly to be so I searched on Google to see if there was a better way to combine observables and it turns out there are a few. I tried zip and forkJoin since they looked the most like what I wanted to do, but nothing would work with them either.

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
});

It could be something as simple as I'm not calling the functions correctly, but I'd have thought I would get some sort of warning or error at some point. tsc doesn't have any problem with the code and I get no messages in the developer consoles on either Chrome or Firefox.

Update

I've tried combineLatest, but it still doesn't display anything in the console. I must be missing something, but I'm not sure what. They work individually.

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

The observables are create in the following way:

...

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 ) );
  });
}

...

Then to actually get the observables I send a listen for responses on the relevant event. e.g.

...

public cassette_designs: Observable<CassetteData[]>;

...

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

Answer 1

To follow up with your findings:

1) An Observable is a lazy execution data type, this means that it will not execute anything in the pipeline until it is subscribed to. This goes for the combinatorial operators as well. zip, forkJoin, combineLatest, and withLatestFrom will all recursively subscribe to the Observables that you pass them only after they themselves have subscriptions.

Hence:

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

Won't actually do anything until you call output.subscribe(), at which point subscribe is also called on stream1 and stream2 and you get all the magic of Rx.

2) A more minor point but whenever you start doing your own creation methods, have a look first at the documentation to see if it already exists. There are static creation methods for, Arrays, Promises, Node-style callbacks and yes even standard event patterns.

Thus your Listen method can become:

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