java - 将 ByteBuffer 流转换为 Rx 中的行的有效方法

Question

我要改造Observable<ByteBuffer>通过按行结束字符拆分成行 ( Observable<String> )。如果我们有类似 toString 的函数, concat和 splitByLine ，我们必须能够做到以下几点:

Observable<ByteBuffer> o = ...;
o.map(toString).reduce(concat).flatMap(splitByLine);

然而，该算法需要先扫描整个字节并将它们存储在内存中，然后才能实际发出反序列化字符串的第一行。如何在每次行尾递增地出现在字节中时发出新行？

Answer 1

我终于找到了rxjava-string .它为我们提供了Rx 运算符 来处理分块 字节数组和字符串流。 API文档找到here .

题目可以通过以下方式实现:

Observable<byte[]> o = ...;
Charset charset = Charset.forName("UTF-8");
// StringObservable have no operators for ByteBuffer yet
StringObservable.byLine(StringObservable.decode(o, charset));