如何将需要多个先前可观测值的依赖订阅链接起来

Question

rxjs的新手，正在寻找一种很好的、干净的方式来链接一些可观测性/订阅。

下面是我制作的一些简单的伪代码，试图模仿我想要做的事情

class Baker implements OnInit {
//functions, constructor, varibles, etc.
ngOnInit(): void {
this.myFridgeService.getDough().pipe(
switchMap(dough => {
return of(this.addToBowl(dough));
}),
switchMap(bowl => {
this.myPantryService.getMnMs()
.subscribe((mnms: Candy[]) => {
this.bagOfMnMs = mnms; // will use later
return of(mnms).pipe(filter(mnm => mnm.isGreen()));
}); 
}),
switchMap(greenOnes => {
return this.myMixerService.mix(bowl, greenOnes); // can I get bowl here somehow?
}))
.subscribe(cookie => {
this.bake(cookie);
},
(error: HttpErrorResponse) => {
console.error(error);
this.serviceError = error;
}
);
}
}

我无法将bowl放入myMixerService，因为它来自以前的switchMap，而且据我所知，我无法传递多个返回。

我可以将myMixerService调用放入myPantryService的订阅事件中，它看起来像这个

class Baker implements OnInit {
ngOnInit(): void {
this.myFridgeService.getDough().pipe(
switchMap(dough => {
return of(this.addToBowl(dough));
}),
switchMap(bowl => {
this.myPantryService.getMnMs()
.subscribe((mnms: Candy[]) => {
this.bagOfMnMs = mnms; // will use later
const greenOnes = of(mnms).pipe(filter(mnm => mnm.isGreen()));
return this.myMixerService.mix(bowl, greenOnes);
},
(error: HttpErrorResponse) => {
console.error(error);
this.serviceError = error;
});
})
).subscribe(cookie => {
this.bake(cookie);
},
(error: HttpErrorResponse) => {
console.error(error);
this.serviceError = error;
}
);
}
}

但我觉得把服务调用放在订阅事件中是一种反模式，事实上这也是switchMap最初的部分原因。除此之外，它永远不会回到cookie

Answer 1

您混合了命令式和反应式风格，这就是您遇到麻烦的原因。您的代码应该在管道中没有任何(this.thing = thing)副作用的情况下工作。相反，定义没有状态和副作用的管道，然后使用异步管道在模板中使用它们。

例如，不要使用这种类型的代码：

@Component({
changeDetection: ChangeDetectionStrategy.Default,
template: '<h2>V: {{ initialValue }} v*2: {{ doubledValue }}</h2>'
})
export class MyComponent implements OnInit {
public initialValue: number;
public doubledValue: number;
ngOnInit() {
this.valueService.getValue().pipe(
tap(it => (this.initialValue = it)), 
map(it => it*2)
).subscribe(it => (this.doubledValue = it));
}
})

相反，做这种风格：

@Component({
changeDetection: ChangeDetectionStrategy.OnPush,
template: '<h2>V: {{ initialValue$ | async }} v*2: {{ doubledValue$ | async }}</h2>'
})
export class MyComponent {
public readonly initialValue$: Observable<number> = this.valueService.getValue();
public readonly doubledValue$: Observable<number> = this.initialValue$.pipe(
map(it => it*2)
);
})

你会得到很多好处：

• OnPush ChangeDetection，效率高得多
• 自动生命周期管理：当模板呈现时，可观察对象会被订阅，当值发生变化时，仅当值发生改变时，模板才会被重新呈现，当组件被破坏时，可观测对象会自动取消订阅(无泄漏(
• 不再有可变状态(根据我的经验，这是错误的头号原因(
• 简洁明了：模板获得的每个值都是单个管道的结果，你知道该在哪里查找
• 防止恶意黑客攻击，例如，邪恶的人再也不可能@ViewChild了——将此组件注入他们的组件中，然后覆盖其值

Answer 2

如果你取任何一个observable，你的subscribe到它，你会得到一个subscription。订阅不是可观察的，也不能转化为可观察的。因此，订阅是一项终端操作。订阅后，你就完成了

简而言之，在switchMap中订阅与您的需求背道而驰。

---

以下是我如何将您的psudo代码重写为应该编译的东西
(假设我正确阅读了您的代码(

this.myFridgeService.getDough().pipe(
map(dough => this.addToBowl(dough)),
switchMap(bowl => 
this.myPantryService.getMnMs<Candy[]>().pipe(
tap(mnms => this.bagOfMnMs = mnms), // will use later
map(mnms => mnms.filter(mnm => mnm.isGreen())),
map(greenOnes => [bowl, greenOnes])
)
),
switchMap(([bowl, greenOnes]) => 
this.myMixerService.mix(bowl, greenOnes)
)
).subscribe({
next: cookie => this.bake(cookie),
error: e =>  {
console.error(e);
this.serviceError = e;
},
complete: () => console.log("Done Baking cookies")

});