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