title: RxJava `switchOnNext` with infinite streams
body: <p>I am trying to write a simple example that uses the <code>switchOnNext</code> operator. I want to generate an infinite stream of infinite streams. EAch individual stream will generate from 1 to infinity. </p>
  
  <p>Using <code>switchOnNext</code> I would expect that each observable will emit it's first <code>n</code> elements, and then the next one, and so on.</p>
  
  <p>To generate an observable that generates values from 1 to infinty I have implemented the static <code>rangeInf</code> function. The <code>main</code> method contains the logic that is supposed to print the values. </p>
  
  <p>When running the program however only the first stream is subscribed to and only its values are printed to the console. For sanity's sake I took the interval observable (built-in and also infinite), but there the behaviour is as expected. </p>
  
  <p>What am I missing here?</p>
  
  <p>At first I figured it would be because interval is on its own seperate thread. But I tried adding <code>.observeOn(Schedulers.computation())</code> in the <code>rangeInf</code> method, but that doesn't seem to fix the problem either.</p>
  
  <h2>Output with <code>interval</code></h2>
  
  <pre><code>&gt; Task :Main.main()
  Next observable
  id 144: value 1
  id 144: value 1
  id 144: value 1
  id 144: value 1
  id 144: value 1
  id 144: value 1
  id 144: value 1
  id 144: value 1
  id 144: value 1
  Next observable
  id 115: value 1
  id 115: value 1
  id 115: value 1
  id 115: value 1
  ...
  </code></pre>
  
  <h2>Output with <code>Observable.generate</code></h2>
  
  <pre><code>&gt; Task :Main.main()
  Next observable
  id 173: value 0
  id 173: value 1
  id 173: value 2
  id 173: value 3
  id 173: value 4
  id 173: value 5
  ...
  </code></pre>
  
  <h2>Source</h2>
  
  <pre><code>import io.reactivex.rxjava3.annotations.NonNull;
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.functions.BiConsumer;
  import io.reactivex.rxjava3.functions.BiFunction;
  import io.reactivex.rxjava3.functions.Consumer;
  import io.reactivex.rxjava3.functions.Supplier;
  import io.reactivex.rxjava3.schedulers.Schedulers;
  
  import java.util.Random;
  import java.util.concurrent.TimeUnit;
  
  public class Main {
      static Random r = new Random();
      public static void main(String[] args) throws InterruptedException {
          Observable&lt;Observable&lt;String&gt;&gt; inf =
                  Observable.interval(0, 10, TimeUnit.SECONDS)
                          .map(x -&gt; {
                              int id = r.nextInt(1000);
                              return Main.rangeInf().map(v -&gt; String.format("%d: %d", id, v));
                          })
                          .doOnNext(i -&gt; System.out.println("Next observable"));
  
          Observable.switchOnNext(inf)
                  .subscribe(System.out::println);
  
          while(true) {
              Thread.sleep(10000);
          }
      }
  
      public static Observable&lt;Integer&gt; inf() {
          Observable&lt;Integer&gt; inf=  Observable.interval(1, TimeUnit.SECONDS)
                  .map(x -&gt; (int) Math.random() * 3000 + 1);
  
          return inf;
  
      }
      public static Observable&lt;Integer&gt; rangeInf() {
          // Initial state.
          Supplier&lt;Integer&gt; s = () -&gt; 0;
          // Generator.
          BiFunction&lt;Integer, Emitter&lt;Integer&gt;, Integer&gt; nxt = (i, e) -&gt; {
              e.onNext(i);
              delay(); // delay random amount of time.
              return i + 1;
          };
          return Observable.generate(s, nxt);
      }
  
      public static void delay() {
          int random = (int) (Math.random() * 1 + 1);
          try {
              Thread.sleep(random * 1000);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
      }
  }
  </code></pre>