RxFirebaseDatabase.java 文件源码

/**
 * Listener for changes in te data at the given query location.
 *
 * @param query    reference represents a particular location in your Database and can be used for reading or writing data to that Database location.
 * @param strategy {@link BackpressureStrategy} associated to this {@link Flowable}
 * @return a {@link Flowable} which emits when a value of the database change in the given query.
 */
@NonNull
public static Flowable<DataSnapshot> observeValueEvent(@NonNull final Query query,
                                                       @NonNull BackpressureStrategy strategy) {
   return Flowable.create(new FlowableOnSubscribe<DataSnapshot>() {
      @Override
      public void subscribe(final FlowableEmitter<DataSnapshot> emitter) throws Exception {
         final ValueEventListener valueEventListener = new ValueEventListener() {
            @Override
            public void onDataChange(DataSnapshot dataSnapshot) {
               emitter.onNext(dataSnapshot);
            }

            @Override
            public void onCancelled(final DatabaseError error) {
               emitter.onError(new RxFirebaseDataException(error));
            }
         };
         emitter.setCancellable(new Cancellable() {
            @Override
            public void cancel() throws Exception {
               query.removeEventListener(valueEventListener);
            }
         });
         query.addValueEventListener(valueEventListener);
      }
   }, strategy);
}