SwitchMap 上的 Forkjoin 在订阅后不返回

Question

我正在编写一个可观察量，它将从服务器获取用户，并试图通过仅获取唯一用户然后缓存它们来提高性能。我还使用 forkjoin 在返回所有用户后更新用户数组。

首先，我使用服务调用从服务器获取用户：

getUsersById(user_ids: number[]): Observable<User[]> {
return Observable.forkJoin(user_ids.map(x => this.getUserById(x)));
}

这将使用以下函数，该函数根据对象是否存在于缓存中或是否需要从服务器检索对象返回对象的可观察量。

public getUserById(user_id: number): Observable<User> {
let tmpUser = this.users.find(x => x.id == user_id);
if (tmpUser != null) {
return Observable.of(tmpUser);
} else {
// Get Token From Cognito Session
return this.cognito.getIdToken()
.flatMap(token => {
// Convert token into Header and retrieve from server
let headers = this.getAuthHeader(token);
return this.http
.get(`${this.userURL}/${user_id}`, {headers: headers})
.map((response) => response.json())
.map(result => {
let newUser = new User();
newUser.deserialize(result);
this.users.push(newUser);
return newUser;
});
});
}
}

调用服务的组件上的代码如下所示：

ngOnChanges() {
if(this.reports.length != 0) {
// Get users for reports
let users = this.reports.map(
x => {
return x.user_id;
});
// Get unique users
let uniqueUsers = users.filter(function (item, i, ar) {
return ar.indexOf(item) === i;
});
// Make call (ForkJoin) and wait for all to return
this.userService.getUsersById(uniqueUsers).subscribe(
users => {
// Add users to user array
this.users = this.reports.map(report => {
return users.find(user => user.id === report.user_id);
});
this.reportsUpdate.emit(this.reports);
}, err => {
console.log(err);
});
}
}

我遇到的问题是 ForkJoin 永远不会返回，即使我可以验证 getUserById 调用是否正在运行完成并返回值。我很困惑，并且已经尝试了开关图和平面图，即使我不完全确定其中的区别。

**编辑**

getIdToken 的代码如下

public getIdToken(): Observable<string> {
return Observable.create(
(observer: Observer<string>) => {
let cognitoUser = this.getCurrentUser();
if (cognitoUser != null) {
cognitoUser.getSession((err, session) => {
if (err) {
console.log(err);
observer.error(err);
} else {
if (session.isValid()) {
observer.next(session.getIdToken().getJwtToken());
}
}
});
} else {
observer.error('Failed To Retrieve Id Token');
}
});
};
getCurrentUser() {
return this.getUserPool().getCurrentUser();
}

我已经确认该程序确实上线了：

observer.next(session.getIdToken().getJwtToken());

因此，观察者未完成似乎不是问题所在

Answer 1

问题出在 cognito.getIdToken() 调用中。我已经编写了在调用时发出令牌值的可观察量。问题是 Observable.next() 函数与 Observable.complete() 不同，我在之后添加了调用，但它不起作用，如下所示

public getIdToken(): Observable<string> {
return Observable.create(
(observer: Observer<string>) => {
let cognitoUser = this.getCurrentUser();
if (cognitoUser != null) {
cognitoUser.getSession((err, session) => {
if (err) {
console.log(err);
observer.error(err);
} else {
if (session.isValid()) {
observer.next(session.getIdToken().getJwtToken());
observer.complete(); // NEW LINE HERE!!!
}
}
});
} else {
observer.error('Failed To Retrieve Id Token');
}
});
};