组间最后一个值的总和

Question

假设我们有一个合作游戏，玩家获得和失去分数。任何玩家都可以随时加入游戏。一旦加入，玩家永远不会离开（为简单起见）。球队的总分只是每个球员在任何时刻得分的总和。玩家应该能够看到自己当前的得分和球队的总得分。

所以，我们有一个绑定到时间戳的所有玩家的分数流。

Time:   0  1  2 3 4 5  6  7    8  9 10 11 12 13 14 15 16    17  18 19 20
Stream: A5 A4       B2 A3 A4B4 A9         K5    B6 A8 B10K9 A12

字母代表球员，数字代表得分。 A4B4 之类的值表示两个玩家的得分同时发生了变化。在groupBy 之后，我们有三个流：每个流用于每个玩家。请注意，玩家的集合不是预定义的。

Time:   0  1  2 3 4 5  6  7    8  9 10 11 12 13 14 15 16    17  18 19 20

Alex:   5  4           3  4    9                   8        12
Ben:                2     4                     6     10
Katie:                                    5           9

我期望得到的是每次分数变化的总和：

Time:   0  1  2 3 4 5  6  7    8  9 10 11 12 13 14 15 16    17  18 19 20
Total:  5  4        6  5  8    13         18    20 19 27    31

那是微不足道的。但解决方案变得复杂。

这是一个输入流。

Observable< Record > input = Observable.from( new Record[] {
  new Record( "Alex", 0, 5 ),
  new Record( "Alex", 1, 4 ),
  new Record( "Ben", 5, 2 ),
  new Record( "Alex", 6, 3 ),
  new Record( "Alex", 7, 4 ),
  new Record( "Ben", 7, 4 ),
  new Record( "Alex", 8, 9 ),
  new Record( "Katie", 12, 5 ),
  new Record( "Ben", 14, 6 ),
  new Record( "Alex", 15, 8 ),
  new Record( "Katie", 16, 9 ),
  new Record( "Ben", 16, 10 ),
  new Record( "Alex", 17, 12 )
} )
.delay( e -> Observable.interval( e.timestamp, TimeUnit.SECONDS, scheduler ) )
.doOnCompleted( latch::countDown )
.share();

记录是一个简单的元组：Record( String player, int timestamp, double score )。

让我们把它分成几组：

Observable< Observable< Record > > output = input
  .groupBy( e -> e.player )
  .map( g -> g.cache() )

标准的toList 方法需要一个 Observable 来完成。但是组流只有在输入流完成时才会完成（任何玩家都可以随时加入游戏）。所以需要一个新的变压器。它在onNext 调用而不是onComplete 返回一个列表。

  .compose( new ToListOnNextTransformer() )

然后，每次发出新组（每次有新玩家加入游戏）时，我们都会生成一个新的 Observable。这个 Observable 返回所有玩家最新得分的总和。

  .map( eachPlayerRecordsList -> Observable.combineLatest( eachPlayerRecordsList, ( Object... eachPlayerRecordsArray ) -> {
    double total = Arrays
      .stream( eachPlayerRecordsArray )
      .mapToDouble( e -> ( ( Record )e ).score )
      .sum();
    int timestamp = Arrays
      .stream( eachPlayerRecordsArray )
      .mapToInt( e -> ( ( Record )e ).timestamp )
      .max()
      .getAsInt();
    return new Record( "Total", timestamp, total );
} ) );

最后，我们输出第一组的值，直到第二个玩家加入游戏。然后我们输出第二组的值，直到第三个玩家加入等等。这正是switchOnNext 所做的。

Observable.switchOnNext( outputg )
  .subscribe( System.out::println, e -> { ( ( Throwable )e ).printStackTrace(); } );

自定义转换器类：

private static class ToListOnNextTransformer implements Observable.Transformer< Observable< Record >, List< Observable< Record > > > {
  private final List< Observable< Record > > list = new ArrayList<>();
  private final PublishSubject< List< Observable< Record > > > ret = PublishSubject.create();

  @Override
  public Observable< List< Observable< Record > > > call( Observable< Observable< Record > > eachPlayerRecords ) {
    eachPlayerRecords.subscribe( playerRecords -> {
      list.add( playerRecords );
      ret.onNext( new ArrayList<>( list ) );
    } );
    return ret;
  }
}

问题是：

• 可以简化吗？最好采用被动式方法。
• 我不是违反了 RxJava 的一些规则吗？比如在订阅者中使用订阅者等。

这是一项人工任务。我只是在学习 RxJava 并试图理解反应式编程。我希望存在更好的方法，但我只是遗漏了一些明显的东西。

The full source code of a test file.

Answer 1

我不完全理解您的示例代码，但您可以通过scan 进行运行求和：

source.scan(0, (a, b) -> a + b)

因此，给定人员及其积分的事件流，您可以创建两个组：

ConnectableObservable<Record> co = input.publish(); // replay, cache, etc.

Observable<Pair<String, Integer>> userScores = co
.groupBy(r -> r.name)
.flatMap(g -> just(Pair.of(g.getKey(), g.scan(0, (a, b) -> a + b))));

Observable<Pair<String, Integer>> teamScores = co
.groupBy(r -> r.team)
.flatMap(g -> just(Pair.of(g.getKey(), g.scan(0, (a, b) -> a + b))));

userScores.filter(p -> p.name.equals("Alex"))
.concatMap(p -> p.second.map(s -> Pair.of(p.first, s)))
.subscribe(System.out::println);

co.connect();