JavaRx：サブスクライバの反応なしにobservableのデフォルト値を設定できますか？

Question

私は、ダウンロードのパーセンテージをソケットからフロントエンドに送るダウンローダを持っています。

DownloadStream downloadStream = SomeService.downloadStream(someFile); 
Long total = downloadStream.totalSize(); 
Subject totalDownloaded = new BehaviorSubject.createDefault(0); 
Disposable downloadSubscriber = totalDownloaded.subscribe(downloaded -> { 
    Message message = new Message(downloaded, total); 
    send(message); 
}); 


Long downloaded = downloadStream.stream(); //stream() is bulk read/write. 
while(downloaded != -1L) { 
    chunk = downloadStream.stream(); 
    totalDownloaded.onNext(totalDownloaded.value + chunk); 
} 

ただし、defaultValue = 0はsubscribeのlambdaを呼び出します。副作用なしにデフォルト値を定義するにはどうすればよいですか？または、長い間観測可能な解決策がいくつかありますか？助けてください...

Answer 1

flatmap（）とfilter（）の機能を使用して値を削除できます。

例：

// create sequence 
    Observable<Integer> o = Observable.just(0, 10, 25, 32, 56, 74, 100); 

    System.out.println("all values:"); 
    o.subscribe(a -> System.out.printf("%d%%; ", a)); 

    System.out.println("\n\nfilter by flatMap:"); 
    o.flatMap(val -> { 
       if(val > 5) 
        return Observable.just(val); 
       else 
        return Observable.empty(); 
      }) 
      .subscribe(a -> System.out.printf("%d%%; ", a)); 

    System.out.println("\n\nfilter by filter():"); 
    o.filter(val -> val > 15) 
      .subscribe(a -> System.out.printf("%d%%; ", a));