如何根据流中的元素动态划分连续的 RXJava 流？

Question

最简单的例子是像这样的字符串流：

["3", "a", "b", "c", "1", "a", "2", "a", "b"]

数字表示它的组应该包含多少个元素。

非常重要的一点是，流是连续的，所以我们不能只是等待下一个数字来分割流。

据我所知，RXJava2中没有内置功能

var flowable = Flowable.concat(Flowable.fromArray("3", "a", "b", "c", "1", "a", "2", "a", "b"), Flowable.never());

flowable/*Something here*/.blockingSubscribe(System.out::println);

预期的输出将是：

[3, a, b, c]

[1, a]

[2, a, b]

Answer 1

我后来发现了阿卡诺克的RxJava2扩展包。使用它，我能够构建这个，它可以做我想要的：var flowable = Flowable.concat(Flowable.fromArray("3", "a", "b", "c", "1", "a", "2", "a", "b"), Flowable.never());flowable.compose(FlowableTransformers.bufferUntil(new Predicate<>() {&nbsp; &nbsp; private int remaining = 0;&nbsp; &nbsp; @Override&nbsp; &nbsp; public boolean test(String next) {&nbsp; &nbsp; &nbsp; &nbsp; if(next.chars().allMatch(Character::isDigit)) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; remaining = Integer.parseInt(next);&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; return --remaining < 0;&nbsp; &nbsp; }})).blockingSubscribe(System.out::println);