RxJava。可読性のあるファイルを読む

Question

私はRxJavaとリアクティブプログラミングにとって全く新しいです。 ファイルを読んでObservableに保存する必要がある課題があります。私は内部でBufferedReaderでCallableを作成しようとしましたが、Observable.fromCallable（）を使用しましたが、うまく動作しませんでした。

どうすればいいですか？

私はRxJava 2.0を使用しています。

Answer 1

this implementationのようなことができます。このように

そしてそのクラスが使用されているhere：

public static Observable<byte[]> from(InputStream is, int size) { 
    return Observable.create(new OnSubscribeInputStream(is, size)); 
} 

は最終的にあなたがそれを使用することができます。

Observable<byte[]> chunks = Bytes.from(file, chunkSize); 

詳細hereを。

Answer 2

私はデータを生成して、オブザーバーが加入するまで観察可能の作成を延期する入れ子になったクラスFileObservableSourceを使用し、基本的なソリューション：

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