title: Combining two Rx stream conditionally
body: <p>I am trying to implement a scenario using Rx where I have two hot Observables. Stream 1 and Stream 2. Based on data of Stream 1 i need to start Stream 2 or Stop Stream 2. Then combine both the stream data into one using CombineLatest. Below id the code that i am able to come up with. </p>
  
  <ol>
  <li><p>Is there a better way to implement this? </p></li>
  <li><p>And How can i make it more generic like I will have Stream 1 and then Stream 2 .. n for each stream from 2 .. n there are condition Condition 2 .. n which utilizes data of Stream 1 to check if other stream needs to start or not and then combine all the data in CombineLatest manner</p></li>
  </ol>
  
  <p><strong>CODE:</strong></p>
  
  <pre><code>        IDisposable TfsDisposable = null;
  
          // stream 1
          var hotObs = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(1));
  
  
          // stream 2
          var hotObs2 = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(1)).Publish();
  
  
          var observerHot =  hotObs.Do(a =&gt; 
          {
              // Based on Condition to start the second stream
              if (ConditionToStartStream2)
              {
                  TfsDisposable = TfsDisposable ?? hotObs2.Connect();
              }
          })
          .Do(a =&gt; 
          {
              // Based on condition 2 stop the second stream
              if (ConditionToStopStream2)
              {
                  TfsDisposable?.Dispose();
                  TfsDisposable = null;
              }
          }).Publish();
  
  
          // Merge both the stream using Combine Latest
          var finalMergedData  = hotObs.CombineLatest(hotObs2, (a, b) =&gt; { return string.Format("{0}, {1}", a, b); });
  
          // Display the result
          finalMergedData.Subscribe(a =&gt; { Console.WriteLine("result: {0}", a);  });
  
          // Start the first hot observable
          observerHot.Connect();
  </code></pre>