使用 RxJava Flowable - 如何根据事件类型拆分发射

Question

假设我们使用 Firebase ChildEventListener，它可以作为多个数据源（它的函数），我用 Flowable 或 Observable 包装它。我希望在每个源中，发射器将数据发送到不同的管道，因为在每种情况下，数据都可以更改，我想以不同的方式处理它——也就是说，根据类型将发射分成几个不同的流事件。

如何在 Java 中做到这一点？

  public void newUsers() {
    DatabaseReference ref = database.getReference().child("Users");
    Flowable.create(emitter -> {
        ref.addChildEventListener(new ChildEventListener() {
            @Override
            public void onChildAdded(@NonNull DataSnapshot dataSnapshot, @Nullable String s) {
                User userData = dataSnapshot.getValue(User.class);
                emitter.onNext(userData);
            }

            @Override
            public void onChildChanged(@NonNull DataSnapshot dataSnapshot, @Nullable String s) {
             User userData = dataSnapshot.getValue(User.class);
                emitter.onNext(userData);
            }

            @Override
            public void onChildRemoved(@NonNull DataSnapshot dataSnapshot) {
             User userData = dataSnapshot.getValue(User.class);
                emitter.onNext(userData);
            }

            @Override
            public void onChildMoved(@NonNull DataSnapshot dataSnapshot, @Nullable String s) {
                emitter.onNext(userData);
            }

            @Override
            public void onCancelled(@NonNull DatabaseError databaseError) {
             User userData = dataSnapshot.getValue(User.class);
                emitter.onNext(userData);
            }
        });
    }
}

Answer 1

因此，要拥有多个流，我们不能只有一个 Flowable。我不知道 Flowable 是否真的是一个要求，所以我将概述我的想法，您可以随意更正它并适应您的需求。

   BehaviourSubject childAddedStream;
   BehaviourSubject childChangedStream;
   BehaviourSubject childRemovedStream;
   BehaviourSubject childMovedStream;
   BehaviourSubject cancelledStream;

  public void newUsers() {
    DatabaseReference ref = database.getReference().child("Users");
    
    ref.addChildEventListener(new ChildEventListener() {
            @Override
            public void onChildAdded(@NonNull DataSnapshot dataSnapshot, @Nullable String s) {
                User userData = dataSnapshot.getValue(User.class);
                childAddedStream.onNext(userData);
            }

            @Override
            public void onChildChanged(@NonNull DataSnapshot dataSnapshot, @Nullable String s) {
             User userData = dataSnapshot.getValue(User.class);
             childChangedStream.onNext(userData);
            }

            @Override
            public void onChildRemoved(@NonNull DataSnapshot dataSnapshot) {
             User userData = dataSnapshot.getValue(User.class);
             childRemovedStream.onNext(userData);
            }

            @Override
            public void onChildMoved(@NonNull DataSnapshot dataSnapshot, @Nullable String s) {
            childMovedStream.onNext(userData);
            }

            @Override
            public void onCancelled(@NonNull DatabaseError databaseError) {
             User userData = dataSnapshot.getValue(User.class);
             cancelledStream.onNext(userData);
            }
        });
    }
}

这样的事情将是我的起点。

这个想法是让这个类为每个回调初始化不同的主题，在创建回调时将回调注册到 FirebaseDatabase，并将各个回调路由到不同的流（主题，请随意使用更相关的主题，而不是 BehaviourSubject，如果你want) - 那么需要监听的人只需监听相关的流而不是单个 Flowable，后者就像回调和反应世界之间的桥梁。