title: Modifying a RxJava stream while it's executing
body: <p>I have an app where users can receive notifications based on topics they subscribe to. The workflow is the following:</p>
  
  <ol>
  <li>User logs in</li>
  <li>The app registers to the notification server</li>
  <li>The users chooses to subscribe/unsubscribe from various topics</li>
  </ol>
  
  <p>I want all the network requests to be serialized. If I knew at initialization what exact topics are going to be subscribed/unsubscribed, I could write a stream like below:</p>
  
  <pre><code>    loginObservable.subscribeOn(Schedulers.io())
              .flatMap(user -&gt; registerApp(appId))
              .flatMap(o -&gt; subscribeToTopic("topic1"))
              .flatMap(o -&gt; unsubscribeFromTopic("topic2"))
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe()
  </code></pre>
  
  <p>The thing is that users can subscribe/unsubscribe at any point in the app lifecycle, possible even before registration is successful. I could maintain a list of observable and serialize all the requests by hand in <code>onComplete()</code> but that doesn't sound very Rx-ish. Any chance I could do it in a more concise way ? Something like:</p>
  
  <pre><code>    observable = loginObservable.subscribeOn(Schedulers.io())
              .flatMap(user -&gt; registerApp(appId))
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe()
  
      //later
      observable.flatMap(o -&gt; subscribeToTopic("topic1"))
              .subscribe()
  
      //even later
      observable.flatMap(o -&gt; unsubscribeFromTopic("topic2"))
              .subscribe()
  </code></pre>