RxFirebaseDatabase.java 文件源码

/**
 * This methods observes a firebase query and returns back
 * an Observable of the {@link DataSnapshot}
 * when the firebase client uses a {@link ValueEventListener}
 *
 * @param firebaseRef {@link Query} this is reference of a Firebase Query
 * @return an {@link rx.Observable} of datasnapshot to use
 */
public Observable<DataSnapshot> observeValueEvent(final Query firebaseRef) {
  return Observable.create(new Observable.OnSubscribe<DataSnapshot>() {
    @Override public void call(final Subscriber<? super DataSnapshot> subscriber) {
      final ValueEventListener listener =
          firebaseRef.addValueEventListener(new ValueEventListener() {
            @Override public void onDataChange(DataSnapshot dataSnapshot) {
              subscriber.onNext(dataSnapshot);
            }

            @Override public void onCancelled(DatabaseError error) {
              FirebaseDatabaseErrorFactory.buildError(subscriber, error);
            }
          });

      // When the subscription is cancelled, remove the listener
      subscriber.add(Subscriptions.create(new Action0() {
        @Override public void call() {
          firebaseRef.removeEventListener(listener);
        }
      }));
    }
  }).compose(this.<DataSnapshot>applyScheduler());
}