@Test public void canGetKeyFromDataSnapshot() { final DataSnapshot expectedKeyData = mock(DataSnapshot.class); when(expectedKeyData.getKey()).thenReturn(EXPECTED_KEY); Query query = mock(Query.class); doAnswer(new Answer<ValueEventListener>() { @Override public ValueEventListener answer(InvocationOnMock invocation) throws Throwable { ValueEventListener valueEventListener = (ValueEventListener) invocation.getArguments()[0]; valueEventListener.onDataChange(expectedKeyData); return valueEventListener; } }).when(query).addValueEventListener(any(ValueEventListener.class)); Observable<String> observable = Observable.create(new ListenToValueEventsOnSubscribe<>(query, getKey())); TestObserver<String> testObserver = new TestObserver<>(); observable.subscribe(testObserver); testObserver.assertValues(EXPECTED_KEY); }
@Override public void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception { final ValueEventListener eventListener = query.addValueEventListener(new RxValueListener<>(emitter, marshaller)); emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(eventListener); } }); }
ListenToValueEventsOnSubscribe implements ObservableOnSubscribe<T> { @Override public void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception { final ValueEventListener eventListener = query.addValueEventListener(new RxValueListener<>(emitter, marshaller)); emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(eventListener); } }); } }
ListenToValueEventsOnSubscribe implements ObservableOnSubscribe<T> { @Override public void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception { final ValueEventListener eventListener = query.addValueEventListener(new RxValueListener<>(emitter, marshaller)); emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(eventListener); } }); } ListenToValueEventsOnSubscribe(Query query, Function<DataSnapshot, T> marshaller); }
ListenToValueEventsOnSubscribe implements ObservableOnSubscribe<T> { @Override public void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception { final ValueEventListener eventListener = query.addValueEventListener(new RxValueListener<>(emitter, marshaller)); emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(eventListener); } }); } ListenToValueEventsOnSubscribe(Query query, Function<DataSnapshot, T> marshaller); @Override void subscribe(@NonNull ObservableEmitter<T> emitter); }
ListenToValueEventsOnSubscribe implements ObservableOnSubscribe<T> { @Override public void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception { final ValueEventListener eventListener = query.addValueEventListener(new RxValueListener<>(emitter, marshaller)); emitter.setCancellable(new Cancellable() { @Override public void cancel() throws Exception { query.removeEventListener(eventListener); } }); } ListenToValueEventsOnSubscribe(Query query, Function<DataSnapshot, T> marshaller); @Override void subscribe(@NonNull ObservableEmitter<T> emitter);  }