使用RxJava从Firebase实时数据库加入扁平化数据

Question

加入扁平化数据是文档中描述的常见用例.但文档显示了一个非实时的简单示例,它不会对更改做出反应.我正在寻找一个更强大的实现.我认为RxJava非常适合这种情况.

考虑遵循Firebase结构:

{
  "messages": {
    "group_id_1": {
      "message_id_1": {
        "text": "Hello",
        "author": "uid_1"
      }
    }
  },
  "users": {
    "uid_1": {
      "name": "David"
    }
  },
  "rooms": {
    "room_id_1": {
      "name": "General",
      "members": {
        "uid_1": true
      }
    }
  }
}

我在这里看到两个用例:

1. 获取具有作者姓名的组中的消息列表
   • 我想我会得到Observable<Message>,当我订阅它时,依赖项(这些消息的用户)也会在某些缓存中订阅.当我显示消息时,我可以从缓存中获取作者的姓名.
   • 它也是实时的 - 如果作者姓名更改,则observable会发出更改的消息.
   • 当我取消订阅observable时,也依赖于unsubscribes.
2. 获取房间成员列表及其名称
   • 我想我会得到Observable<User>,当我订阅它时,它将首先订阅房间的成员,然后订阅个人用户.
   • 它是实时的 - 如果房间成员发生变化,我会收到通知.
   • 当我取消订阅observable时,也依赖取消订阅.

你知道库/解决方案可以做到吗？

或者如果我创建一个,你会使用它吗？

Answer 1

我本来打算提出这个问题的一个变体，但似乎在这个问题的基础上构建可能会更好......我将描述希望至少部分回答上述问题的内容，但也是我的一个缺点看到。

使用上面的数据模型，我们可能会像下面这样围绕 firebase 查询创建 RxJava 包装器，以获取特定房间的成员键列表并获取特定成员的详细信息（注意使用onCompleted()in subscriber.getMemberInfo...稍后会详细介绍！）。

public Observable<String> getRoomMembers(String roomId) {
    return Observable.create(subscriber -> {
        databaseReference.child(roomId + "/members").addValueEventListener(new ValueEventListener() {
            @Override
            public void onDataChange(DataSnapshot dataSnapshot) {
                for (DataSnapshot childSnapshot: dataSnapshot.getChildren()) {
                    String userId = childSnapshot.getKey()
                    subscriber.onNext(userId);
                }
                subscriber.onCompleted();
            }

            @Override
            public void onCancelled(DatabaseError databaseError) {
            }
        });
    });
}

public Observable<Member> getMemberInfo(String memberId) {
    return Observable.create(subscriber -> {
        databaseReference.child(memberId).addValueEventListener(new ValueEventListener() {
            @Override
            public void onDataChange(DataSnapshot dataSnapshot) {
                Member member = dataSnapshot.getValue(Member.class);
                subscriber.onNext(member);
                subscriber.onCompleted();
            }

            @Override
            public void onCancelled(DatabaseError databaseError) {
            }
        });
    });
}

然后我们可以使用类似以下内容来获取Member特定房间的列表（已添加isActive属性以Member显示我们如何过滤获得的结果）。

    getRoomMembers(roomId)
            .flatMap(memberId -> getMemberInfo(memberId))
            .filter(Member::isActive)
            .toList()
            .subscribe(members -> {

            });

所以，上述内容在一定程度上是有效的。问题是我必须调用subscriber.onCompleted()上述getMemberInfo调用才能flatMap工作......这意味着对Member数据的任何后续更改都不会触发上述订阅中的更新。我相对较新RxJava，Firebase因此可能会遗漏一些明显的东西。