根据多个可观察量处理数据

Question

我有两个来自两个不同HTTP调用的可观察量。 在我的组件中，我需要调用这两个服务，处理从这些服务返回的数据，并将另外两个变量发送到模板，这两个变量是处理的结果。

我已经尝试了很多我无法将它们全部放在这里的事情，但这是我迄今为止尝试过的主要解决方案：

• 在 2 个服务上使用 forkJoin，然后处理数据：执行该过程的函数表示结果未定义。
• 使用开关映射：与上述结果相同。
• 使用行为主题：再次获得相同的结果

我不知道我是否忘记了 Observable 或 RxJs 的一些基本规则，但我真的找不到正确的方法来做到这一点。

这是我在组件中的当前代码：

public notifGroups: any;
public currentUser: any;
public currentUserGroups: any;
public notifGroupsImOut: any;
private _notifGroupsImIn: BehaviorSubject<any> = new BehaviorSubject([]);
public readonly notifGroupsImIn: any = this._notifGroupsImIn.asObservable();
constructor(
private _groupService: GroupService,
private _currentUserService: CurrentUserService,
private _userService: UserService,
) { }

ngOnInit() {
this.test().subscribe(
res => {
console.log(res);
for(let mGroup in res[1]["user"]["subscriptionsToGroups"]){
for(let nGroup in res[0]["list"]) {
console.log(nGroup);
console.log(mGroup);
if (mGroup["group"]["email"] == nGroup["email"]){             
this._notifGroupsImIn
.next(this._notifGroupsImIn.getValue().push(nGroup));
}
}
}
this.notifGroupsImIn.subscribe()
}
)
}

private _getNotifGroups(): any{
return this._groupService
.getGroupsByType('NOTIFICATION')
}
private _getCurrentUser() {
return this._currentUserService
.getCurrentUser()  
}
public getAll(){
return Observable.forkJoin(this._getNotifGroups(), this._getCurrentUser())
}
public test(): Observable<Array<any>> {
let obs = this.getAll();
obs.subscribe(
res => {
console.log(res);
this.notifGroups = res[0]["list"];
this.currentUser = res[1]["user"]["subscriptionsToGroups"];
}
)
return obs
}

好吧，我知道代码的某些部分现在很脏，但我希望你不要挖太多。

所以我要做的是：

我从两个服务中得到以下值：通知组和当前用户组。

它们都是对象列表。

然后在我的test((函数中，我将每个对象与另一个对象进行比较，如果该对象的电子邮件属性等于另一个对象的电子邮件属性，则将其添加到notifGroupImIn中。

但是当调用test((函数时，看起来http调用的结果尚未解决，即使我在ngOnInit中调用它们并订阅了它们。

我不确定我是否足够容易理解，所以请告诉我。

谢谢。

更新

我已经更改了一些代码，因为在模板中，notifGroupsImIn仍然是一个可观察的。现在我有以下错误：无法读取未定义的属性"电子邮件"。引发错误的电子邮件属性是以下两个属性之一：

if (mGroup["group"]["email"] == nGroup["email"]){                  
this._notifGroupsImIn.next(this._notifGroupsImIn.getValue().push(nGroup));
}

另外，两条线

console.log(nGroup);
console.log(mGroup);

两者都在控制台中返回 0，无法弄清楚原因，因为res[1]["user"]["subscriptionsToGroups"]和res[0]["list"]正确返回对象列表

Answer 1

forkJoin的简单示例

@Component({
selector: 't-root',
template: `
<div>{{groups}}</div>
<div>{{user}}</div>
`
})
export class AppComponent {
groups: any;
user: any;
// Assume this method calls this._groupService.getGroupsByType('NOTIFICATION')
private _getNotifGroups(): Observable<any> {
return Observable.of('GROUPS');
}
// Assume this method calls this._currentUserService.getCurrentUser()  
private _getCurrentUser(): Observable<any> {
return Observable.of('USER');
}
private getAll(): Observable<any> {
return Observable.forkJoin( this._getNotifGroups(), this._getCurrentUser() );
}
ngOnInit(): void {
this.getAll().subscribe(res => {
console.log(res); // ['GROUPS', 'USER']
this.groups = res[0];
this.user = res[1];
});
}
}

Answer 2

你试过combineLatest吗

combineLatest：每当任何可观察序列生成元素时，使用选择器函数将指定的可观察序列合并为一个可观察序列。

public getAll(){
return Observable.combineLatest(
this._getNotifGroups(), 
this._getCurrentUser()
);
}

combineLatest文档

如果它不起作用，请说明错误是什么或为什么它不起作用。