RXJS等待其他可观察

Question

我需要订阅的结果，而是等待中间业务，以获取结果之前完成。诀窍是，我“访问”我的结果来填充它：

// a service that gets a model
service.getModel(): Observable<MyModel>;

// I need to enrich my model before consuming it
service.getModel()
    .makeSureAllCodesAreFetched(data => visitModel(model))
    .subscribe(data => console.log("data is ready: ", data));

// a visitor that visits the model tree and enriches the leaves
// recursively visit the branches
visitModel(model: MyModel) {
    if (model.isLeaf) {
       // on condition, call a service to fetch additional data
       service.fetchCodes(model.codeKey).subscribe(codes => model.codes = codes);
    } else {
        model.properties.forEach(prop: MyModel => visit(prop));
    }
}

我试着用合并和forkJoin（）没有成功发挥。我只是想确保所有来电fetchCodes()，无论结果如何，都做在我的数据订阅。

Answer 1

我发现了一个解决方案，但它不是最干净的，以我的意见。

// a service that gets a model
service.getModel(): Observable<MyModel>;

// I need to enrich my model before consuming it
service.getModel()
    .pipe(
        mergeMap(data => forkJoin(visitModel(model))))
    .subscribe(data => console.log("data is ready: ", data[0]));

// a visitor that visits the model tree and enriches the leaves
// recursively visit the branches
visitModel(model: MyModel, obs?: Observable<MyModel>[]): Observable<MyModel>[] {
    if (obs === undefined) {
        obs = [];
        obs.push(of(model)); // make sure the very first Observable is the root
    }
    if (model.isLeaf) {
       // on condition, call a service to fetch additional data
       // push Observable result in the array
       obs.push(service.fetchCodes(model.codeKey).map(codes => {
           model.codes = codes;
           return model;
       }));
    } else {
        model.properties.forEach(prop: MyModel => visit(prop, obs)); // recursive call
    }
    return obs;
}

我的游客将真正追加所有来电fetchCodes()到Observables数组并返回它。这样forkJoin将等待所有的呼叫完成。诀窍（和肮脏的部分），我必须确保第一Observable实际上是根元素，我感兴趣的东西。