Angular 2 - Observable chain and operations

Question

So i'm quite new to angular and also to observables in general. I know how to deal with promises but I would like to be more comfortable with the Observable side.

I'll try to keep it short but here's the setup:

In order to reduce the amount of calls made to the server, I created a package service that basically packages what would be multiple requests into a single one.

A component would ask multiple data services to refresh their store if their cache aren't up to date. Every data services that needs an update from the server adds a request to the package. When ready, the component sends the package to the package service which in turn will send the request to the server.

When the server replies with data, I need every data services to update their store with the new data pertaining to their single requests and then notify the component that everything has completed.

Here's how I would do it with promises :

1) Component builds a package from every needed data services.

2) Component sends it to the package service that returns a promise.

3) The package service sends the request(s).

4) When the server replies, the package service dispatches the response to every data services registered in the package.

5) The package service resolves its promise only when every involved data services are done updating their stores.

6) The component deals with the resolved/rejected promise.

So trying to convert this to an observable pattern I get this:

Component:

let thePackage = this.packageService.createPackageInstance();

this.service1.loadData( thePackage );
this.service2.loadData( thePackage );
this.packageService.sendPackage( thePackage ).subscribe( () => console.log("Success"), (err) => console.error( err ) );

Service1 and Service2:

loadData( thePackage: Package ) {

    let needUpdate = this.GetCacheIsUpToDate();

    if( needUpdate ) {

        let params = { 
          "param1": true
        };

        //This would return a promise in a promise chain pattern.
        let callback = ( ( response:any ) => {

          //[...]
          //Do update the data store and notify subscribers.

          return Promise.resolve(); //The only way I know to make it work.
        });

        this.packageService.createRequest( thePackage, 100, 1, params, callback );
    }
}

PackageService:

createPackageInstance(): Package {

  return new Package({/* Some params */});
}


createRequest( thePackage:Package, theModule:number, operation:number, params:any, callback:Function ):void {

  let data = JSON.stringify( params );

  let request = new Request({
    "id":         Utils.generateGUID(),
    "theModule":  theModule,
    "operation":  operation,
    "data" :      data,
    "callback":   callback
  });

  thePackage.requests.push( request );
}


sendPackage( thePackage ): Observable<PackageResponse>{

  //Assume some variables like 'someUrl' to be setup
  //[...]

  let httpObs = this.http.post( someUrl, formData, someOptions ).map( response => {

    return response.json();
  });

  return httpObs.flatMap( ( response ) => {

     let pAll = [];
     thePackage.requests.forEach( request => {
         pAll.push( request.callback( response ) );
     });

     return Observable.fromPromise( Promise.all( pAll ) ); //This is really ugly
  });
}

EDIT: Edited some code to show my current version that works with promises.

And what's the result of your code? What is your question? Is it working? If not, do you have any hint where/why it fails? -- Edit: Ok sorry, I haven't read inside the code... — olivarra1

olivarra1 olivarra1 · Accepted Answer · 2017-06-29T15:57:48

Sorry if this answer is not complete enough - I don't have much time to go into too much detail, but it's too long to fit inside a comment.

The way you have it now, you need to get the original callback you defined in createRequest for each of the data coming from the server. You should have a similar method for your promise-based solution, right?. But that is not too much Rx...

What I would do is make createRequest return an observable where the service can subscribe to. Inside createRequest this observable under-the-hood would be a Rx.BehaviorSubject: This is an observable that's an observer as well, so every time you call .next(message) on it, the subscriptors of that subject will receive that message. Then store this subject in the package.

And then, instead of calling a callback after you make the request, you'd call .next() on the subject you just stored with the relevant data requested. I'd call .complete() inmediately after, so the stream gets disposed (assuming we receive no more that for that stream, which seems the case)

Keep in mind that Rx.BehaviourSubjects are discouraged to be used, because normally there's a better way to handle those cases, but in this case I can't find a better option (maybe there is, but I'm not sure)

Angular 2 - Observable chain and operations

1 Answers