java - RxJava。读取文件以观察

Question

我对 RxJava 和响应式编程完全陌生。我有一个任务，我必须读取文件并将其存储到 Observable 中。我尝试用 BufferedReader 制作一个 Callable 并使用 Observable.fromCallable() ，但效果不佳。

你能告诉我该怎么做吗？

我正在使用 RxJava 2.0。

Answer 1

一个基本解决方案，我使用嵌套类 FileObservableSource 来生成数据，然后推迟 Observable 的创建，直到观察者订阅:

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

public class StackOverflow {

    public static void main(String[] args) {
        final Observable<String> observable = Observable.defer(() -> new FileObservableSource("pom.xml"));
        observable.subscribe(
                line -> System.out.println("next line: " + line),
                        Throwable::printStackTrace,
                        () -> System.out.println("finished")
                );
    }

    static class FileObservableSource implements ObservableSource<String> {

        private final String filename;

        FileObservableSource(String filename) {
            this.filename = filename;
        }

        @Override
        public void subscribe(Observer<? super String> observer) {
            try {
                Files.lines(Paths.get(filename)).forEach(observer::onNext);
                observer.onComplete();
            } catch (IOException e) {
                observer.onError(e);
            }
        }
    }
}