将 Observable 的 Observable 转换为简单的 Observable

Question

我对 rxjs 很陌生，已经阅读了几十篇教程，但仍然很困惑。 假设我有一个用户 ID 列表，每 3 秒我想从 REST 服务查询每个用户的一些数据（在线游戏比赛信息）。多个用户可能玩过相同的比赛，因此 getLastMatch() 可能会为不同的用户返回相同的比赛 id。我想按匹配 ID 对结果流进行分组。 因此，对于用户 1，我得到匹配 id 101、2 -> 100 和 3 -> 101。 我希望我的 observable 发出类似

的东西

{
    [
    {"match": 101, "players": [1, 3]},
    {"match": 100, "players": [2]},
    ]
}

这是到目前为止我想出的代码，但我坚持使用最后一行生成 可观察>

class MatchData{
  constructor(public matchId: number, public won: boolean) {}
}

const players = [1, 2, 3];

function getLastMatch(userId: number): Observable<MatchData> {
  let data = new MatchData(100 + userId % 2, userId % 2 == 0);
  return Observable.fromPromise(new Promise<MatchData>(resolve => resolve(data))));
}

const scheduler = Observable.interval(3000);
scheduler.map(Observable.from(players.map(p => getLastMatch(p))));

UPD：

这就是我最终的结果。

class MatchData {
  constructor(public playerId: number, public matchId: number) {}
}

class GroupedMatchData {
  constructor(public matchId: number, public playerIds: number[]) {}
}

const accounts = [1, 2, 3];
const scheduler = Observable.interval(3000);

function getMatch(id: number): Observable<MatchData> {
  let promise = new Promise<MatchData>(resolve => resolve(new MatchData(id, 100 + id % 2)));
  return Observable.fromPromise(promise);
}

function requestMatchData(): Observable<GroupedMatchData> {
  return Observable.from(accounts.map(account => getMatch(account)))
    .mergeAll()
    .groupBy(match => match.matchId, match => match.playerId)
    .flatMap(group => group.reduce((accumulator, current) => [...accumulator, current], [group.key]))
    .map(array => new GroupedMatchData(array[0], array.slice(1)));
} 

scheduler.take(1).flatMap(requestMatchData).subscribe(console.log);

感谢我的解决方案中的任何 cmets。

Answer 1

我们可以使用Observable.forkJoin 将玩家列表映射到他们最近的比赛列表。

使用 flatMap 我们可以摆脱嵌套的 Observables。

因为forkJoin 将按照与玩家相同的顺序返回匹配项，我们可以将每个（玩家，匹配）对合并为单个对象。

然后我们可以通过 matchId 对结果进行分组。

   const scheduler = Observable.interval(3000);
    scheduler
      .flatMap(() => Observable.forkJoin(...players.map(id => getLastMatch(id))))
      .map(mergeMatchesWithPlayers)
      .map(groupPlayersByMatch)
      .subscribe(console.log)

    function mergeMatchesWithPlayers(matches: MatchData[]) {
      return matches.map((match, i) => ({player: players[i], match}));
    }

    function groupPlayersByMatch(pairs: {player: number, match: MatchData}[]) {
      const groups = [];
      pairs.forEach(pair => {
          const existingGroup = groups.find(group => group.match === pair.match.matchId);
          existingGroup 
            ? existingGroup.players.push(pair.player)
            : groups.push({match: pair.match.matchId, players: [pair.player]})
      });
      return groups;
    }

https://codepen.io/anon/pen/ZrPLRb