title: Combining Firebase realtime data listener with RxJava
body: <p>I am using <a href="https://www.firebase.com/" rel="noreferrer">Firebase</a> in my app, along with RxJava.
  Firebase is capable of notify your app whenever something changed in the backend data (addition, removals, changes, ...).
  I am trying to combine the feature of Firebase with RxJava.</p>
  
  <p>The data I am listening for is called <code>Leisure</code>, and the <code>Observable</code> emits <code>LeisureUpdate</code> which contains a <code>Leisure</code> and the type of update (add, remove, moved, changed).</p>
  
  <p>Here is my method which allows to subscribe to this events.</p>
  
  <pre><code>private Observable&lt;LeisureUpdate&gt; leisureUpdatesObservable;
  private ChildEventListener leisureUpdatesListener;
  private int leisureUpdatesSubscriptionsCount;
  
  @NonNull
  public Observable&lt;LeisureUpdate&gt; subscribeToLeisuresUpdates() {
      if (leisureUpdatesObservable == null) {
          leisureUpdatesObservable = Observable.create(new Observable.OnSubscribe&lt;LeisureUpdate&gt;() {
  
              @Override
              public void call(final Subscriber&lt;? super LeisureUpdate&gt; subscriber) {
                  leisureUpdatesListener = firebase.child(FirebaseStructure.LEISURES).addChildEventListener(new ChildEventListener() {
                      @Override
                      public void onChildAdded(DataSnapshot dataSnapshot, String s) {
                          final Leisure leisure = convertMapToLeisure((Map&lt;String, Object&gt;) dataSnapshot.getValue());
                          subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.ADDED));
                      }
  
                      @Override
                      public void onChildChanged(DataSnapshot dataSnapshot, String s) {
                          final Leisure leisure = convertMapToLeisure((Map&lt;String, Object&gt;) dataSnapshot.getValue());
                          subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.CHANGED));
                      }
  
                      @Override
                      public void onChildRemoved(DataSnapshot dataSnapshot) {
                          final Leisure leisure = convertMapToLeisure((Map&lt;String, Object&gt;) dataSnapshot.getValue());
                          subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.REMOVED));
                      }
  
                      @Override
                      public void onChildMoved(DataSnapshot dataSnapshot, String s) {
                          final Leisure leisure = convertMapToLeisure((Map&lt;String, Object&gt;) dataSnapshot.getValue());
                          subscriber.onNext(new LeisureUpdate(leisure, LeisureUpdate.MOVED));
                      }
  
                      @Override
                      public void onCancelled(FirebaseError firebaseError) {
                          subscriber.onError(new Error(firebaseError.getMessage()));
                      }
                  });
              }
          });
      }
      leisureUpdatesSubscriptionsCount++;
      return leisureUpdatesObservable;
  }
  </code></pre>
  
  <p>First off, I would like to use <code>Observable.fromCallable()</code> method in order to create the <code>Observable</code>, but I guess it is impossible, since Firebase uses callbacks, right?</p>
  
  <p>I keep a single instance of the <code>Observable</code> in order to always have one <code>Observable</code> where multiple <code>Subscriber</code> can subscribe.</p>
  
  <p>The problem comes when everyone unsubscribe and I need to stop listening for the events in Firebase.
  I didn't find anyway to make the <code>Observable</code> understand if there is any subscription still. So I keep counting how many calls I got to <code>subscribeToLeisuresUpdates()</code>, with <code>leisureUpdatesSubscriptionsCount</code>.</p>
  
  <p>Then every time someone wants to unsubscribe it has to call</p>
  
  <pre><code>@Override
  public void unsubscribeFromLeisuresUpdates() {
      if (leisureUpdatesObservable == null) {
          return;
      }
      leisureUpdatesSubscriptionsCount--;
      if (leisureUpdatesSubscriptionsCount == 0) {
          firebase.child(FirebaseStructure.LEISURES).removeEventListener(leisureUpdatesListener);
          leisureUpdatesObservable = null;
      }
  }
  </code></pre>
  
  <p>This is the only way I found to make the <code>Observable</code> emits items when there is a subscriber, but I feel like there must be an easier way, specially understanding when there is no more subscribers listening to the observable.</p>
  
  <p>Anyone who encountered a similar problem or have a different approach? </p>