Эффективный способ преобразования потока 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 для обработки потока chunked байтовые массивы и строки. Документ API найден here.

Вопрос может быть достигнуто следующим образом:

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