Angular 2 - 可观察的链和操作

Question

所以我对 Angular 和一般的 observables 都很陌生。我知道如何处理 Promise，但我希望更适应 Observable 方面。

我会尽量保持简短，但设置如下：

为了减少对服务器的调用量，我创建了一个包服务，它基本上将多个请求打包成一个请求。

如果缓存不是最新的，组件会要求多个数据服务刷新其存储。每个需要从服务器更新的数据服务都会向包中添加一个请求。准备好后，组件将包发送到包服务，包服务反过来将请求发送到服务器。

当服务器回复数据时，我需要每个数据服务使用与其单个请求相关的新数据更新其存储，然后通知组件一切都已完成。

这就是我如何使用 promises 来做的：

1) 组件从每个需要的数据服务构建一个包。

2) 组件将其发送到返回承诺的包服务。

3) 包服务发送请求。

4) 当服务器回复时，包服务将响应分派给包中注册的每个数据服务。

5) 包服务只有在每个涉及的数据服务完成更新其存储时才履行其承诺。

6) 组件处理已解决/已拒绝的承诺。

所以尝试将其转换为可观察的模式，我得到了：

组件：

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

包服务：

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

编辑：编辑了一些代码以显示我当前的适用于 Promise 的版本。

Answer 1

对不起，如果这个答案不够完整 - 我没有太多时间详细介绍，但是太长了，无法放入评论中。

按照您现在的方式，您需要为来自服务器的每个数据获取您在createRequest 中定义的原始callback。您应该为基于承诺的解决方案使用类似的方法，对吧？但这不是太多的 Rx...

我要做的是让createRequest 返回一个服务可以订阅的可观察对象。在 createRequest 内部，这个 observable 底层将是一个 Rx.BehaviorSubject：这是一个 observable 也是一个观察者，所以每次你调用 .next(message) 时，该主题的订阅者都会收到该消息。然后将此主题存储在包中。

然后，您无需在发出请求后调用回调，而是在刚刚存储的主题上调用.next() 并请求相关数据。之后我会立即致电.complete()，因此流被处理（假设我们不再收到该流的消息，情况似乎如此）

请记住，不鼓励使用Rx.BehaviourSubjects，因为通常有更好的方法来处理这些情况，但在这种情况下我找不到更好的选择（也许有，但我没有当然）