title: How To Have Multiple Subscribers To A Grouped Observable in RxJava?
body: <p>This is using RxJava version <strong>0.19.6</strong>.</p>
  
  <p>Outside of a groupBy operation, one can create a pipeline described by the following code to, for instance, select a record from an Observable based on some criteria or select the first record that meets some alternate criteria:</p>
  
  <pre><code>Observable&lt;Long&gt; observable = Observable.interval(1, TimeUnit.MILLISECONDS).take(10);
  Observable&lt;Long&gt; filter1 = observable.filter(new Func1&lt;Long, Boolean&gt;() {
      @Override
      public Boolean call(Long aLong) {
          return 5 == aLong % 5;
      }
  });
  Observable&lt;Long&gt; filter2 = observable.filter(new Func1&lt;Long, Boolean&gt;() {
      @Override
      public Boolean call(Long aLong) {
          return 2 == aLong % 5;
      }
  });
  BlockingObservable.from(Observable.concat(filter1, filter2).first()).forEach(new Action1&lt;Long&gt;() {
      @Override
      public void call(Long aLong) {
          System.out.println(aLong);
      }
  });
  </code></pre>
  
  <p>...unfortunately, due to restrictions on the GroupedObservable, it appears that the same kind of procedure does not work will operating inside of a grouped context:</p>
  
  <pre><code>BlockingObservable.from(observable.groupBy(new Func1&lt;Long, Long&gt;() {
      @Override
      public Long call(Long aLong) {
          return aLong % 5;
      }
  }).flatMap(new Func1&lt;GroupedObservable&lt;Long, Long&gt;, Observable&lt;Long&gt;&gt;() {
      @Override
      public Observable&lt;Long&gt; call(GroupedObservable&lt;Long, Long&gt; in) {
          Observable&lt;Long&gt; filter1 = in.filter(new Func1&lt;Long, Boolean&gt;() {
              @Override
              public Boolean call(Long aLong) {
                  return 5 == aLong % 5;
              }
          });
          Observable&lt;Long&gt; filter2 = in.filter(new Func1&lt;Long, Boolean&gt;() {
              @Override
              public Boolean call(Long aLong) {
                  return 2 == aLong % 5;
              }
          });
          return Observable.concat(filter1, filter2).first();
      }
  })).forEach(new Action1&lt;Long&gt;() {
      @Override
      public void call(Long aLong) {
          System.out.println(aLong);
      }
  });
  </code></pre>
  
  <p>...results in a multiple subscriber exception (Exception in thread "main" java.lang.IllegalStateException: Only one subscriber allowed!).</p>
  
  <p>Am I missing some obvious fix to this problem?  I have tried playing around with ConnectableObservables in this case to give the appearance of a single Subscriber, but those attempts have been failures as well (surely due to ignorance on my part).</p>
  
  <hr>
  
  <p>On a related note, the groupByUntil seems to give you a reference to the GroupedObservable as well which was giving me a similar headache of complaining about multiple subscribers if I actually tried to use it to determine when to close the window.  Here again I'm sure that I am overlooking something obvious since the API clearly expects one to use the GroupedObservable!</p>