RxJava wrapper for Firebase ValueEventListeners

Query.java

package com.usehomeroom.vasuki.data;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fernandocejas.arrow.optional.Optional;
import com.google.firebase.database.DataSnapshot;
import com.google.firebase.database.DatabaseError;
import com.google.firebase.database.DatabaseReference;
import com.google.firebase.database.FirebaseDatabase;
import com.google.firebase.database.ValueEventListener;

import rx.Observable;
import rx.Subscriber;
import rx.subscriptions.Subscriptions;

/**
 * Created by Vinay on 13/05/16.
 */
public class Query<T> implements Observable.OnSubscribe<DataSnapshot> {
    private final TypeReference<T> typeIndicator;
    private final ObjectMapper mapper = new ObjectMapper();

    private DatabaseReference firebase;

    public Query(TypeReference<T> typeIndicator) {
        this.typeIndicator = typeIndicator;
    }

    public Observable<Optional<T>> execute(final String path) {
        firebase = FirebaseDatabase.getInstance().getReference(path);
        return Observable.create(this)
                .map(this::parseSnapshot);
    }

    @Override public void call(Subscriber<? super DataSnapshot> subscriber) {
        ValueEventListener listener = new ValueEventListener() {
            @Override public void onDataChange(DataSnapshot dataSnapshot) {
                subscriber.onNext(dataSnapshot);
            }

            @Override public void onCancelled(DatabaseError databaseError) {
                subscriber.onError(databaseError.toException());
            }
        };
        firebase.addValueEventListener(listener);
        subscriber.add(Subscriptions.create(() -> firebase.removeEventListener(listener)));
    }

    private Optional<T> parseSnapshot(DataSnapshot snapshot) {
        if (snapshot.exists()) {
            return Optional.of(mapper.convertValue(snapshot.getValue(), typeIndicator));
        } else {
            return Optional.absent();
        }
    }
}

SingleQuery.java

package com.usehomeroom.vasuki.data;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fernandocejas.arrow.optional.Optional;
import com.google.firebase.database.DataSnapshot;
import com.google.firebase.database.DatabaseError;
import com.google.firebase.database.DatabaseReference;
import com.google.firebase.database.FirebaseDatabase;
import com.google.firebase.database.ValueEventListener;

import rx.Observable;
import rx.Subscriber;
import rx.subscriptions.Subscriptions;

/**
 * Created by Vinay on 21/05/16.
 */
public class SingleQuery<T> implements Observable.OnSubscribe<DataSnapshot> {

    private final TypeReference<T> typeIndicator;
    private final ObjectMapper mapper = new ObjectMapper();

    private DatabaseReference firebase;

    public SingleQuery(TypeReference<T> typeIndicator) {
        this.typeIndicator = typeIndicator;
    }

    public Observable<Optional<T>> execute(final String path) {
        firebase = FirebaseDatabase.getInstance().getReference(path);
        return Observable.create(this)
                .map(this::parseSnapshot);
    }

    @Override public void call(Subscriber<? super DataSnapshot> subscriber) {
        ValueEventListener listener = new ValueEventListener() {
            @Override public void onDataChange(DataSnapshot dataSnapshot) {
                subscriber.onNext(dataSnapshot);
                subscriber.onCompleted();
            }

            @Override public void onCancelled(DatabaseError databaseError) {
                subscriber.onError(databaseError.toException());
            }
        };
        firebase.addListenerForSingleValueEvent(listener);
        //Not sure about removing single value event listeners
        subscriber.add(Subscriptions.create(() -> firebase.removeEventListener(listener)));
    }

    private Optional<T> parseSnapshot(DataSnapshot snapshot) {
        if (snapshot.exists()) {
            return Optional.of(mapper.convertValue(snapshot.getValue(), typeIndicator));
        } else {
            return Optional.absent();
        }
    }
}