RxFirebase.java 文件源码

@NonNull public static Observable<DataSnapshot> singleValue(final Query query) {
    return Observable.create(new Observable.OnSubscribe<DataSnapshot>() {
        @Override public void call(final Subscriber<? super DataSnapshot> subscriber) {
            final ValueEventListener valueEvent = query.addValueEventListener(new ValueEventListener() {
                @Override public void onDataChange(DataSnapshot dataSnapshot) {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onNext(dataSnapshot);
                    }
                }
                @Override public void onCancelled(DatabaseError databaseError) {
                    subscriber.onError(databaseError.toException());
                }
            });
            subscriber.add(Subscriptions.create(new Action0() {
                @Override public void call() {
                    query.removeEventListener(valueEvent);
                }
            }));
        }
    });
}