Rxjava + Firebase

public class MainActivity extends RxAppCompatActivity {

    private static final String TAG = MainActivity.class.getSimpleName();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        DatabaseReference ref = FirebaseDatabase.getInstance().getReference().child("-KRVOI4q-D4dzp-s8zEs");
        observeValueEvent(ref)
                .flatMap(snapshot -> Observable.just(Note.deserialiseNote(snapshot)))
                .compose(bindToLifecycle())
                .subscribe(this::logOnNext);
    }

    public Observable<DataSnapshot> observeValueEvent(final Query query) {
        return Observable.fromAsync(new Action1<AsyncEmitter<DataSnapshot>>() {
            @Override
            public void call(final AsyncEmitter<DataSnapshot> dataSnapshotAsyncEmitter) {
                final ValueEventListener valueEventListener = query.addValueEventListener(
                        new ValueEventListener() {
                            @Override
                            public void onDataChange(DataSnapshot dataSnapshot) {
                                dataSnapshotAsyncEmitter.onNext(dataSnapshot);
                            }

                            @Override
                            public void onCancelled(final DatabaseError error) {
                                dataSnapshotAsyncEmitter.onError(new FirebaseException(error.getMessage()));
                            }
                        });

                dataSnapshotAsyncEmitter.setCancellation(() -> query.removeEventListener(valueEventListener));
            }
        }, AsyncEmitter.BackpressureMode.BUFFER);
    }

    public void logOnNext(Note note){
        Log.d(TAG, note.toString());
    }
}