RxJava 缓冲区/窗口，直到元素遭受条件

Question

我正在尝试从阅读器中读取行并将它们分组到属于一起的块。

原文：

bla1
bla2
### block separator ###
bla3
bla4
### block separator ###
...

我需要得到两个块（bla1，bla2）和（bla3，bla4）。

代码：

import org.apache.commons.lang3.StringUtils;
import rx.Observable;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.Reader;
import java.util.Iterator;

public class BlockBuilder {

  public static void main(String[] args) {

    try {
      FileReader fileReader = new FileReader("/path/to/some/file");
      LineIterable lineIterable = new LineIterable(fileReader);

      Observable.from(lineIterable)
          .buffer(100)
          // Needed instead of time/count: until line matches condition
          // something like .buffer(line -> line.equals("### block separator ###")
          .forEach(gatheredLines -> {
            String gatheredBlock = StringUtils.join(gatheredLines, '\n');
            System.out.println(gatheredBlock);
            System.out.println("###### ###### ###### ######");
          });
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }

  private static class LineIterable implements Iterable<String> {
    private final Iterator<String> iterator;
    public LineIterable(Reader reader) {
      iterator = new BufferedReader(reader).lines().iterator();
    }
    @Override
    public Iterator<String> iterator() {
      return iterator;
    }
  } 
}

使用缓冲区或窗口无关紧要，或者我对这两者的看法完全错误。

我认为缓冲区的 bufferClosingSelector 或窗口的 closingSelector 一定是可能的。 两者都是创建 Observer 的函数，它可以触发关闭当前缓冲区或窗口，但我看不到在哪里可以获取当前行。

Answer 1

您可以发布您的源代码并将其用于缓冲和缓冲区边界：

Observable<String> source = Observable.just(
        "a", "b", "#", 
        "c", "d", "e", "#", 
        "f", "g");

source.publish(p -> 
        p.filter(v -> !"#".equals(v))
        .buffer(() -> p.filter(v -> "#".equals(v))))
.subscribe(System.out::println);