RxJava 2 - 非同期呼び出しを連鎖させる方法

Question

最近私はRxJavaの学習を始めましたので、初心者の質問をしている場合は私を十字架にしないでください。私が見つけることができるほぼすべてのドキュメントを読みました。ほとんどのチュートリアルではhttp://reactivex.io/tutorials.htmlに従っています。私はStackOverflowとその他のインターネットの高低を調べましたが、明らかに私はこの問題を持つ唯一の人物です。本質的に、ソフトウェアのあらゆる部分が何かをしなければならないので、これは奇妙です。

私が見つけたすべてのチュートリアルは、ストリームにいくつかの機能を適用して、有用で素晴らしい、新しいストリームを作成しています。私の場合は間違っても本当に役に立たないようにしてください。それはまた私に考えさせました...多分私はこれについてすべて間違っています。しかし、私は今深く詰まっているし、マントラに続いて、 "すべてがストリームなんだ"ということはなぜできないのだろう？だからここ

は私がやろうとしているものです：

1. はロード
2. コールCompletable作成を実行するためにSingleを呼び出して、いくつかのサーバー
3. 上のログインアクションを実行するためのいくつかのフォームを示しています
4. Single呼び出しの結果を次のアクションで呼び出し、読み込みを非表示にする一部のサーバーでのユーザーアクション

私はAndroidでこれで終わりますが、達成したいことを概説するための基本的なJava 8の例を作成しました。

これは私がこれまでに思い付いたものです：

注：

• 私はgetMainStream()機能がいくつかの相互作用ここで

をシミュレートすることがありrxjava2

を使用しています実行可能なコードのバージョン：

public static void main(final String[] args) { 
    getMainStream() 
     .doOnNext(__ -> showLoading()) 
     .flatMap(__ -> loginUser().toObservable()) 
     .flatMap(__ -> createUser().toObservable()) 
     .doOnNext(userId -> { 
      hideLoading(); 
      System.out.println("userId: " + userId); 
     }) 
     .subscribe(); 
} 

public static Completable loginUser() { 
    return Completable.create(new CompletableOnSubscribe() { 
     @Override 
     public void subscribe(final CompletableEmitter e) throws Exception { 
      Thread.sleep(500); 
      System.out.println("loginUser"); 
      e.onComplete(); 
     } 
    }); 
} 

public static Single<String> createUser() { 
    return Single.<String>create(new SingleOnSubscribe<String>() { 
     @Override 
     public void subscribe(final SingleEmitter<String> e) throws Exception { 
      Thread.sleep(1000); 
      System.out.println("createUser"); 
      e.onSuccess("some_user_id"); 
     } 
    }); 
} 

public static Completable getCompletable(final String input) { 
    return Completable.create(new CompletableOnSubscribe() { 
     @Override 
     public void subscribe(final CompletableEmitter e) throws Exception { 
      Thread.sleep(750); 
      System.out.println("completable, input=" + input); 
      e.onComplete(); 
     } 
    }); 
} 

public static Observable<Object> getMainStream() { 
    return Observable.just(new Object()); 
} 

private static void hideLoading() { 
    System.out.println("hideLoading()"); 
} 

private static void showLoading() { 
    System.out.println("showLoading()"); 
} 

こののコンソール出力は次のようになります。

showLoading() 
loginUser 

は残念ながら、ログインユーザは返すことはありません！

私は本当にこのトピックに関する助けを楽しみにしています！

ありがとうございました！

Answer 1

loginUser()はCompletableであり、あなたがObservableにCompletableを変換する際に、効果が観察できるが完了することです。したがって、createUser()によって処理される下流の値はありません。

loginUser().andThen(() -> createUser().toObservable()に式を変更すると、文字列が出力されることがあります。