java - Rx - 按条件将流分成段(列表)

Question

我有一个 RX 生产者，它创建一个字符串流(真实流的简化版本):

A1 A2 A3 B1 B2 C1 C2 C3 C4 C5 C6....

川流不息，却又有序。所以以A开头的字符串用完后，B开始。当B用完时，C开始...当Z用完时，我们移动到AA1等。有未知数量的 A、B 等，但通常每个字母有 10-30 个实例。

我正在寻找一种方法将此流分成所有 A 的 block :A1 A2 A3，所有 B 的:B1 B2，所有 C 的: C1 C2 C3 C4 C5 C6 等。每个 block 可以是一个可观察对象(我将把它变成一个列表)或只是一个列表。

我使用 RxJava 尝试了几种不同的方法，但都失败了。无效的部分包括:

• 分组依据:由于流是无止境的，因此每个字母的可观察对象不会完成。所以当 A 用完而 B 开始时，A 的 Observable 没有完成。因此，可观测值的数量不断增加。

• Window/Buffer with distinctUntilChanged - 我在原始流上使用“distinctUntilChanged”来输出每组的第一项(第一个 A、第一个 B 等)。然后我将该流用作 window 的输入或用作窗口/缓冲区之间边界的“缓冲区”运算符。那没有用，我得到的只是空列表。

使用 RX 的正确解决方案是什么？我更喜欢 Java 解决方案，但也非常欢迎可以轻松转换为 Java 的其他 RX 实现中的解决方案。

Answer 1

您可以使用 rxjava-extras .toListWhile:

Observable<String> source = 
    Observable.just("A1", "A2", "A3", "B1", "B2", "B3", "C1", "D1");
source.compose(Transformers.<String> toListWhile(
            (list, t) -> list.isEmpty() 
                         || list.get(0).charAt(0) == t.charAt(0)))
      .forEach(System.out::println);

它做了@akarnokd 在幕后所做的事情并且经过了单元测试。