RxJava.读取文件即可观察 [英] RxJava. Read file to observable

问题描述

我对RxJava和反应式编程完全陌生. 我有一个作业，我必须在其中读取文件并将其存储到Observable.我试图在内部使用BufferedReader进行Callable并使用Observable.fromCallable()，但效果不佳.

I am totally new to RxJava and reactive programming. I have an assignment where i must read file and store it to Observable. I have tried to make a Callable with BufferedReader inside and use Observable.fromCallable(), but it didn't work much.

您能告诉我我该怎么做吗?

Could you please show me how can i do that?

我正在使用RxJava 2.0.

I am using RxJava 2.0.

推荐答案

一种基本解决方案，其中我使用嵌套类FileObservableSource生成数据，然后将Observable的创建推迟到Observer订阅:

A basic solution where I use a nested class FileObservableSource to produce the data and then defer the creation of the Observable until an Observer subscribes:

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);
            }
        }
    }
}

这篇关于RxJava.读取文件即可观察的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！