Combining Firebase realtime data listener with RxJava
<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>