RxJava-файл и оператор цепочки

Question

Я пытаюсь реактивно хвост лог-файл с помощью RxJava-File:

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-файл и ваш тестовый случай, как сообщается, работает от rxjava-файл 0.3.3 на Maven Central.