title: Waiting for parallel RX subscribers to complete
body: <p>I'm looking for the best method to wait for async tasks to finish in rx-java.</p>
  
  <p>As a common example say there is a function which gets a list of id's from a local store and then queries a remote system for those Id's, the remote system results are then consolidated into a single report and returned to the caller of the function. As the call to the remote system is slow we want them to be done in asynchronously and I only want to return once all of the calls have returned and their results have been processed. </p>
  
  <p>The only reliable way I have found to do this is to poll the subscription to check it is unsubscribed yet. But I'm thinking doesn't seem to be the 'RX' way to do things!</p>
  
  <p>As an example I've taken the example from <a href="http://howrobotswork.wordpress.com/2013/10/28/using-rxjava-in-android/" rel="noreferrer">http://howrobotswork.wordpress.com/2013/10/28/using-rxjava-in-android/</a> and amended it slightly to make it non-android and to show what I mean. I have to have the following code at the of the main() method to stop it exiting immediately.</p>
  
  <pre><code>while (!subscription.isUnsubscribed()) {
      Thread.sleep(100);
  }
  </code></pre>
  
  <p>The full code for the example is listed below (it is dependent on <a href="http://square.github.io/retrofit/" rel="noreferrer">http://square.github.io/retrofit/</a> if your trying to compile it)</p>
  
  <pre><code>package org.examples;
  
  import retrofit.RestAdapter;
  import retrofit.http.GET;
  import retrofit.http.Query;
  import rx.Observable;
  import rx.Subscriber;
  import rx.Subscription;
  import rx.functions.Action0;
  import rx.functions.Action1;
  import rx.functions.Func1;
  import rx.schedulers.Schedulers;
  
  public class AsyncExample {
  
      public static void main(String[] args) throws InterruptedException {
          final Subscription subscription = Observable.from("London", "Paris", "Berlin")
                  .flatMap(new Func1&lt;String, Observable&lt;WeatherData&gt;&gt;() {
                      @Override
                      public Observable&lt;WeatherData&gt; call(String s) {
                          return ApiManager.getWeatherData(s);
                      }
                  })
                  .subscribe(
                          new Action1&lt;WeatherData&gt;() {
                              @Override
                              public void call(WeatherData weatherData) {
                                  System.out.println(Thread.currentThread().getName() + " - " + weatherData.name + ", " + weatherData.base);
                              }
                          },
                          new Action1&lt;Throwable&gt;() {
                              @Override
                              public void call(Throwable throwable) {
                                  System.out.println(Thread.currentThread().getName() + " - ERROR: " + throwable.getMessage());
                              }
                          },
                          new Action0() {
                              @Override
                              public void call() {
                                  System.out.println(Thread.currentThread().getName() + " - COMPLETED");
                              }
                          }
                  );
  
          // Have to poll subscription to check if its finished - is this the only way to do it?
          while (!subscription.isUnsubscribed()) {
              Thread.sleep(100);
          }
      }
  }
  
  class ApiManager {
  
      private interface ApiManagerService {
          @GET("/weather")
          WeatherData getWeather(@Query("q") String place, @Query("units") String units);
      }
  
      private static final RestAdapter restAdapter = new RestAdapter.Builder()
              .setEndpoint("http://api.openweathermap.org/data/2.5")
              .build();
      private static final ApiManagerService apiManager = restAdapter.create(ApiManagerService.class);
  
      public static Observable&lt;WeatherData&gt; getWeatherData(final String city) {
          return Observable.create(new Observable.OnSubscribe&lt;WeatherData&gt;() {
              @Override
              public void call(Subscriber&lt;? super WeatherData&gt; subscriber) {
                  try {
                      System.out.println(Thread.currentThread() + " - Getting " + city);
                      subscriber.onNext(apiManager.getWeather(city, "metric"));
                      subscriber.onCompleted();
                  } catch (Exception e) {
                      subscriber.onError(e);
                  }
              }
          }).subscribeOn(Schedulers.io());
      }
  }
  </code></pre>