実行中のRxJavaストリームの変更

Question

私は、ユーザが購読しているトピックに基づいて通知を受け取ることができるアプリを持っています。ワークフローは以下の通りです：

、ユーザーがさまざまなトピック

私が欲しいから退会/サブスクライブすることを選択した通知サーバへ

アプリレジスタの

1. ユーザーログをすべてのネットワーク要求直列化される。私は/解除加入しようとしている、私は以下のような流れを書くことができ、正確なものをトピック初期化時に知っていた場合：

   loginObservable.subscribeOn(Schedulers.io()) 
         .flatMap(user -> registerApp(appId)) 
         .flatMap(o -> subscribeToTopic("topic1")) 
         .flatMap(o -> unsubscribeFromTopic("topic2")) 
         .observeOn(AndroidSchedulers.mainThread()) 
         .subscribe() 
   

   事は、ユーザーがさえ/サブスクライブ・アプリケーションのライフサイクルの任意の時点で解除可能ということです登録が成功する前に私は観測可能なリストを維持することができ、onComplete()の手ですべてのリクエストをシリアライズすることができますが、これはRx-ishのようには聞こえません。私はもっ​​と簡潔な方法でそれをやり遂げることができますか？次のようなもの：

   observable = loginObservable.subscribeOn(Schedulers.io()) 
         .flatMap(user -> registerApp(appId)) 
         .observeOn(AndroidSchedulers.mainThread()) 
         .subscribe() 
   
       //later 
       observable.flatMap(o -> subscribeToTopic("topic1")) 
         .subscribe() 
   
       //even later 
       observable.flatMap(o -> unsubscribeFromTopic("topic2")) 
         .subscribe() 
   

Answer 1

doOnSubscribeはどうなりますか？トピックが公開される前にアプリを登録できる場所。

この例を見てください。

boolean onSubscribe = false; 

@Test 
public void observableDoOnSubscribe() { 
    String val = "test"; 
    Observable.just(val) 
       .doOnSubscribe(() -> onSubscribe = true) 
       .filter(s -> onSubscribe) 
       .subscribe(s -> System.out.printf(s)); 
} 

あなたはPublishSubjectを経て、観察に新しいトピックを導入することができ、ここでより多くの例https://github.com/politrons/reactive

Answer 2

を見ることができます。私はあなたの例ではswitchMapの使用を理解していないが、このアプローチは、あなたに役に立つかもしれない：

PublishSubject<Topic> newTopics = PublishSubject.create(); 
Observable<Topic> topics = ...; 
newTopics 
    .mergeWith(topics) 
    .flatMap(topic -> subscribeToTopic(topic)) 
    ... 
    .subscribe(subscriber); 

UIの相互作用：

newTopics.onNext(topic);