title: Get data asynchronously and then then process data serially with RxJava
body: <p>I have a function that creates an observable that is responsible for fetching some data. This function is called multiple times so that there are N threads in the background that fetch data. Once the data have been fetched, I have to process them serially. Each round of processing should begin when a notification arrives and all the data are processed one by one (we don't care about the order or if there are still X background threads that still fetch data while we process the data we currently have).</p>
  <p>Currently I have a solution implemented but it's not entirely correct as the currentSubject only returns the first value.</p>
  <p>Example code here:</p>
  <pre><code>package Test;
  
  import java.util.concurrent.ThreadLocalRandom;
  import java.util.concurrent.TimeUnit;
  
  import io.reactivex.rxjava3.core.Observable;
  import io.reactivex.rxjava3.schedulers.Schedulers;
  import io.reactivex.rxjava3.subjects.PublishSubject;
  
  public class Test {
    public static void main(String[] args) throws InterruptedException {
      var t = new Test();
  
      var&lt;String&gt; currentSubject = PublishSubject.create();
      var n = 4;
      for (var i = 0; i &lt; n; ++i) {
        Observable.zip(
                currentSubject,
                t.getData(String.valueOf(i), ThreadLocalRandom.current().nextInt(1, 5)),
                (s1, s2) -&gt; s1 + &quot; for &quot; + s2)
            .first(&quot;Default&quot;)
            .doOnEvent(
                (s, e) -&gt; {
                  System.out.println(&quot;Processing result: &quot; + s);
                })
            .subscribe();
      }
  
      for (var i = 0; i &lt; n; ++i) {
        Thread.sleep(i * 2000);
        currentSubject.onNext(&quot;A notification: &quot;+i);
      }
      System.out.println(&quot;End&quot;);
    }
  
    public Observable&lt;String&gt; getData(String s, int sec) {
      return Observable.just(s)
          .delay(sec, TimeUnit.SECONDS)
          .doOnNext(
              dt -&gt; {
                System.out.println(&quot;Got data: &quot; + dt);
              })
          .subscribeOn(Schedulers.newThread());
    }
  }
  </code></pre>
  <p>What am I doing wrong and &quot;currentSubject&quot; only returns the first value? I'm creating multiple Observables each time, can this cause a problem e.g. having multiple Observables in memory while they should have been disposed? And finally how can I use the &quot;currentSubject&quot; multiple times without a problem?</p>