title: RXJava 2 called twice
body: <p>Im still trying to use RXJava 2 for a polling service used by multiple subscribers. It works well except that it always called twice.</p>
  
  <p>I tried using publish(1), take(1), share(), refCount(), etc but the results are always the same. Called twice or multiple time.</p>
  
  <p>I still dont get it why its called twice. New subscribers should receive the latest emitted values and after only changed values (if the hashmap changed) returned as modified list.</p>
  
  <p>My polling observable (Singleton)</p>
  
  <pre><code>public Observable&lt;List&lt;Light&gt;&gt; lightPolling = CallFactory
              .with(context)
              .getLights(MainApplication.lastaccespoint)
              .repeatWhen(o -&gt; o.concatMap((Function&lt;Object, ObservableSource&lt;?&gt;&gt;) v -&gt; Observable.timer(1000, TimeUnit.MILLISECONDS)))
              .filter(new MapPredicate&lt;&gt;())
              .distinctUntilChanged()
  
              .map(l -&gt; {
                  List&lt;Light&gt; lights = new ArrayList&lt;&gt;();
                  for (Map.Entry&lt;String, Light&gt; entry : l.entrySet()) {
                      Light light = entry.getValue();
                      light.setId(entry.getKey());
                      lights.add(light);
                  }
                  return lights;
              })
              // .replay(1)
              //    .distinct()
              //.publish()
              //.share()
              .compose(ReplayingShare.instance()); // its like .share(), just including the last result etc.
  </code></pre>
  
  <p>My Predicate for filtering if the hashmap is really changed. StringUtil.equalMap is just a simple method which compares two hashmaps which works well.</p>
  
  <pre><code> class MapPredicate&lt;T&gt; implements Predicate&lt;Map&lt;String, T&gt;&gt; {
  
      private Map&lt;String, T&gt; lastMap;
  
      @Override
      public boolean test(Map&lt;String, T&gt; map) throws Exception {
          if (!StringHelper.equalMap(map, lastMap)) {
              Timber.d("Result in MapPredicate doesnt equals last result");
              lastMap = map;
              return false;
          }
          return true;
      }
  }
  </code></pre>
  
  <p>Finally my Disposable which subscribes on the Observable.</p>
  
  <pre><code>lightsDisposable = LightManager
              .getInstance(context)
              .lightPolling
              .subscribeWith(new BulbsObserver());
  </code></pre>
  
  <p>Some more for testing</p>
  
  <pre><code> LightManager.getInstance(context).lightPolling.debounce(10, TimeUnit.SECONDS).subscribe(lights -&gt; Timber.d("Received light second Subscriber "+ lights.size()));
   LightManager.getInstance(context).lightPolling.debounce(15, TimeUnit.SECONDS).subscribe(lights -&gt; Timber.d("Received light Third Subscriber "+ lights.size()));
  </code></pre>
  
  <p>And of course my Observer</p>
  
  <pre><code>private class BulbsObserver extends DisposableObserver&lt;List&lt;Light&gt;&gt; {
  
      @Override
      public void onNext(List&lt;Light&gt; newLights) {
          Timber.d("Received lights=" + newLights.size());
          lights.clear();
          lights.addAll(newLights);
          adapter.notifyDataSetChanged();
      }
  
      @Override
      public void onError(Throwable throwable) {
          Timber.e(throwable);
      }
  
      @Override
      public void onComplete() {
          Timber.d("onComplete called");
      }
  }
  </code></pre>
  
  <p>Replaying Share is from Jack Wharton <a href="https://github.com/JakeWharton/RxReplayingShare" rel="nofollow noreferrer">https://github.com/JakeWharton/RxReplayingShare</a></p>