以角度 5 同步嵌套订阅

Question

我有一个类似的功能：

function(param: any): Subject<any> {

    let newsubj: Subject<any> = new Subject<any>();
    let thing;

    this.dataContextService.dataContext.getFullThing({ param: param }).subscribe(result => {
        if (result) {

            thing = result.thing;

            this.dataContextService.dataContext.Table.Query(query => query
                .orderBy(["ID desc"])
                .top(1)
            ).subscribe(number => {
                if (number) {

                    let increment = number + 1;

                    let newObject = new Object({ id: increment, thing: thing });
                    this.dataContextService.dataContext.Favorite.Post(newObject).subscribe(result => {
                        newsubj.next(newObject);
                        newsubj.complete();
                    })

                }

            })
        }

    })
    return newsubj;
}

我无法将这个 http 调用的执行与 rxjs 同步，请有人帮忙吗？ （这里是 rxjs 新手）。谢谢

Answer 1

一种方法是使用forkJoin。它的工作方式类似于Promise.all([...])，但在Observable 字段中。按以下方式修改您的代码：

forkJoin(
        this.dataContextService.dataContext.getFullThing({ param: param }),
        this.dataContextService.dataContext.Table.Query(query => query
            .orderBy(["ID desc"])
            .top(1)
        )
    ).subscribe(
        ([fullThing, number]) => {
            thing = fullThing && fullThing.thing;

            if (number) {
                let increment = number + 1;

                let newObject = new Object({ id: increment, thing: thing });
                this.dataContextService.dataContext.Favorite.Post(newObject).subscribe(result => {
                    newsubj.next(newObject);
                    newsubj.complete();
                });
            }
        }
    );

UPD：根据 cmets 的建议，最好使用 forkJoin 而不是 ForkJoinObservable，因此我编辑了答案以反映这一点。您还可以在此处查看更多forkJoin 使用示例：https://www.learnrxjs.io/operators/combination/forkjoin.html