RxJava-文件和操作符链接

Question

我正在尝试使用RxJava-文件来响应跟踪日志文件

    File file = new File(".\\server.log");
    Observable<String> newLines =
            FileObservable.tailer()
                    .file(file)
                    .startPosition(file.length())
                    .sampleTimeMs(1000)
                    .chunkSize(8192)
                    .utf8()
                    .tailText();

    newLines.subscribe(System.out::println);

它的工作就像预期的那样。但是，一旦我试图将更多的运营商连锁，我就会遇到问题。例如，更改为

  newLines.filter(LogfileWatcher::error).subscribe(System.out::println);

(其中error()是一个简单的函数String -> Boolean)我只在第一个附加到文件之后才得到输出，而不是后面的那个。在使用window()或其他几个运算符时也会出现类似的问题。

我做错了什么？

Answer 1

必须在rxjava-file中修复背压支持，并且据报告您的测试用例在Maven Central上的rxjava-file 0.3.3中工作。