java - 根据判别索引创建多个源

Question

我有一个流，它收集数据作为其受体的响应。

因此，假设我有 n 个受体，在接收到的帧数据中，我能够知道它是从哪个受体索引接收的，并执行过滤器并为每个受体创建一个新源，如下所示。

-----a1-----b1-----c1----a2-----a3----c2----b2-

--------a1-----------------a2--------a3-------------

----------------b1------------------------------b2 -

----------------------c1----------------c2-------- -

下一步包括从每对最新的值中获取

这是我的代码示例:

 data.publish().refCount()
    .groupBy(frame -> frame.getIndex())
    .flatMap(Observable::toList)
    .buffer(2,1)
    .subscribe( delays -> {
        System.out.println(delays);
    });

框架就是 POJO:

public class Frame {
    private int index;
    private String message;
    private long timestamp;

    public Frame(int index, String message) {
        this.index = index;
        this.message = message;
        setTimestamp(System.currentTimeMillis());
    }
}

如何为每个组创建源？然后逐对取出每组的项目

Answer 1

如果您已经打印了给定输入集示例的预期输出，那就太好了，因为您对最新对含义的解释是可以解释的。因此，如果我假设最新的 {a1, a2, a3} 对:(a1, a2) 和 (a2, a3) 那么以下代码将执行您想要的操作:

  private static Observable<List<String>> observeGrouped(Observable<String> str, String group) {
    return str.filter(s->s.substring(0, 1).equals(group)).buffer(2,1).filter(list->list.size() == 2);
  }


  @Test
  public void test() throws InterruptedException {

    Observable<String> src = Observable.just("a1", "b1", "c1", "a2", "a3", "c2", "b2").publish().refCount();

    String[] groups = new String[] {"a", "b", "c"};
    CountDownLatch latch = new CountDownLatch(groups.length);
    for (String group : groups) {
      observeGrouped(src, group).subscribe(System.out::println, e->{}, ()->latch.countDown());
    }
    latch.await();
  }

输出:

[a1, a2]
[a2, a3]
[b1, b2]
[c1, c2]

上面的代码只是一个展示通用方法的原型(prototype)。您可以将类似的逻辑应用于您的特定 POJO。如果您还有其他意思，请告诉我。