@Test public void testObserveValueEvent() { TestSubscriber<ChildData> testObserver = RxFirebaseDatabase .observeValueEvent(databaseReference, ChildData.class) .test(); ArgumentCaptor<ValueEventListener> argument = ArgumentCaptor.forClass(ValueEventListener.class); verify(databaseReference).addValueEventListener(argument.capture()); argument.getValue().onDataChange(dataSnapshot); testObserver.assertNoErrors() .assertValueCount(1) .assertValueSet(Collections.singletonList(childData)) .assertNotComplete() .dispose(); }
@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) { if (!emitter.isCancelled()) emitter.onError(new RxFirebaseDataException(error)); } }; emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(valueEventListener); } }); query.addValueEventListener(valueEventListener); } }, strategy); }
RxFirebaseDatabase { @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) { if (!emitter.isCancelled()) emitter.onError(new RxFirebaseDataException(error)); } }; emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(valueEventListener); } }); query.addValueEventListener(valueEventListener); } }, strategy); } }
RxFirebaseDatabase { @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) { if (!emitter.isCancelled()) emitter.onError(new RxFirebaseDataException(error)); } }; emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(valueEventListener); } }); query.addValueEventListener(valueEventListener); } }, strategy); }  }
RxFirebaseDatabase { @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) { if (!emitter.isCancelled()) emitter.onError(new RxFirebaseDataException(error)); } }; emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(valueEventListener); } }); query.addValueEventListener(valueEventListener); } }, strategy); }  @NonNull static Flowable<DataSnapshot> observeValueEvent(@NonNull final Query query,
                                                           @NonNull BackpressureStrategy strategy); @NonNull static Maybe<DataSnapshot> observeSingleValueEvent(@NonNull final Query query); @NonNull static Single<DataSnapshot> runTransaction(@NonNull final DatabaseReference ref,
                                                      @NonNull final boolean fireLocalEvents,
                                                      @NonNull final long transactionValue); @NonNull static Completable setValue(@NonNull final DatabaseReference ref,
                                       final Object value); @NonNull static Completable updateChildren(@NonNull final DatabaseReference ref,
                                             @NonNull final Map<String, Object> updateData); @NonNull static Flowable<RxFirebaseChildEvent<DataSnapshot>> observeChildEvent(
        @NonNull final Query query, @NonNull BackpressureStrategy strategy); @NonNull static Flowable<DataSnapshot> observeMultipleSingleValueEvent(@NonNull DatabaseReference... whereRefs); @NonNull static Maybe<DatabaseReference[]> requestFilteredReferenceKeys(@NonNull final DatabaseReference from,
                                                                          @NonNull Query whereRef); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Class<T> clazz,
                                                    @NonNull BackpressureStrategy strategy); @NonNull static Maybe<T> observeSingleValueEvent(@NonNull final Query query,
                                                       @NonNull final Class<T> clazz); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Class<T> clazz,
        @NonNull BackpressureStrategy strategy); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Function<? super DataSnapshot, ? extends T> mapper,
                                                    @NonNull BackpressureStrategy strategy); @NonNull static Maybe<T> observeSingleValueEvent(@NonNull final Query query,
                                                       @NonNull final Function<? super DataSnapshot, ? extends T> mapper); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Function<? super RxFirebaseChildEvent<DataSnapshot>,
        ? extends RxFirebaseChildEvent<T>> mapper, @NonNull BackpressureStrategy strategy); @NonNull static Flowable<DataSnapshot> observeValueEvent(@NonNull final Query query); @NonNull static Flowable<RxFirebaseChildEvent<DataSnapshot>> observeChildEvent(
        @NonNull final Query query); @NonNull static Single<DataSnapshot> runTransaction(@NonNull final DatabaseReference ref,
                                                      @NonNull final long transactionValue); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Class<T> clazz); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Class<T> clazz); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Function<? super DataSnapshot, ? extends T> mapper); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Function<? super RxFirebaseChildEvent<DataSnapshot>,
        ? extends RxFirebaseChildEvent<T>> mapper); }
RxFirebaseDatabase { @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) { if (!emitter.isCancelled()) emitter.onError(new RxFirebaseDataException(error)); } }; emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(valueEventListener); } }); query.addValueEventListener(valueEventListener); } }, strategy); }  @NonNull static Flowable<DataSnapshot> observeValueEvent(@NonNull final Query query,
                                                           @NonNull BackpressureStrategy strategy); @NonNull static Maybe<DataSnapshot> observeSingleValueEvent(@NonNull final Query query); @NonNull static Single<DataSnapshot> runTransaction(@NonNull final DatabaseReference ref,
                                                      @NonNull final boolean fireLocalEvents,
                                                      @NonNull final long transactionValue); @NonNull static Completable setValue(@NonNull final DatabaseReference ref,
                                       final Object value); @NonNull static Completable updateChildren(@NonNull final DatabaseReference ref,
                                             @NonNull final Map<String, Object> updateData); @NonNull static Flowable<RxFirebaseChildEvent<DataSnapshot>> observeChildEvent(
        @NonNull final Query query, @NonNull BackpressureStrategy strategy); @NonNull static Flowable<DataSnapshot> observeMultipleSingleValueEvent(@NonNull DatabaseReference... whereRefs); @NonNull static Maybe<DatabaseReference[]> requestFilteredReferenceKeys(@NonNull final DatabaseReference from,
                                                                          @NonNull Query whereRef); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Class<T> clazz,
                                                    @NonNull BackpressureStrategy strategy); @NonNull static Maybe<T> observeSingleValueEvent(@NonNull final Query query,
                                                       @NonNull final Class<T> clazz); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Class<T> clazz,
        @NonNull BackpressureStrategy strategy); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Function<? super DataSnapshot, ? extends T> mapper,
                                                    @NonNull BackpressureStrategy strategy); @NonNull static Maybe<T> observeSingleValueEvent(@NonNull final Query query,
                                                       @NonNull final Function<? super DataSnapshot, ? extends T> mapper); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Function<? super RxFirebaseChildEvent<DataSnapshot>,
        ? extends RxFirebaseChildEvent<T>> mapper, @NonNull BackpressureStrategy strategy); @NonNull static Flowable<DataSnapshot> observeValueEvent(@NonNull final Query query); @NonNull static Flowable<RxFirebaseChildEvent<DataSnapshot>> observeChildEvent(
        @NonNull final Query query); @NonNull static Single<DataSnapshot> runTransaction(@NonNull final DatabaseReference ref,
                                                      @NonNull final long transactionValue); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Class<T> clazz); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Class<T> clazz); @NonNull static Flowable<T> observeValueEvent(@NonNull final Query query,
                                                    @NonNull final Function<? super DataSnapshot, ? extends T> mapper); @NonNull static Flowable<RxFirebaseChildEvent<T>> observeChildEvent(
        @NonNull final Query query, @NonNull final Function<? super RxFirebaseChildEvent<DataSnapshot>,
        ? extends RxFirebaseChildEvent<T>> mapper);  }