使用 RxJava 连接来自 Firebase 实时数据库的扁平化数据

Question

加入扁平化数据是一个常见用例，也在documentation 中进行了描述。但是文档显示了一个简单的示例，它不是实时的，它不会对更改做出反应。我正在寻找更强大的实现。我认为 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 时，依赖项也会取消订阅。
2. 获取房间成员列表及其姓名
   • 我想我会收到Observable<User>，当我订阅它时，它会先订阅房间的成员，然后再订阅个人用户。
   • 它是实时的 - 如果房间成员发生变化，我会收到通知。
   • 当我取消订阅 observable 时，依赖项也会取消订阅。

您知道可以做到这一点的库/解决方案吗？

如果我创建了一个，你会使用它吗？

Answer 1

我最终用这两种方法解决了（在 Kotlin 中，Java 类似，只是更冗长）：

fun <A, B> Observable<List<A>>.mapSubQueries(subQuery: (A) -> Observable<B>): Observable<List<Observable<B>>> {
    return this.flatMap {
        if (it.isNotEmpty()) {
            return@flatMap Observable.from(it).map { subQuery(it) }.toList()
        } else {
            return@flatMap Observable.just(listOf<Observable<B>>())
        }
    }
}

@Suppress("UNCHECKED_CAST")
fun <T> Observable<List<Observable<T>>>.joinSubQueries(): Observable<List<T>> {
    return this.flatMap {
        if (it.isNotEmpty()) {
            return@flatMap Observable.combineLatest(it, {
                val list = mutableListOf<T>()
                it.forEach {
                    list.add(it as T)
                }
                list
            })
        } else {
            return@flatMap Observable.just(listOf<T>())
        }
    }
}

要在所有消息中获取用户，我可以这样使用它：

fun usersInMessages(roomId): Observable<List<User>> {
    return DatabaseRead.messages(roomId)
            .mapSubQueries { DatabaseRead.user(it.getAuthor()) }
            .joinSubQueries()
}

我决定最好将此代码保留在我的代码库中，并针对各种用例稍作修改。使它成为一个库会降低它的灵活性。要点是始终使用Observable.combineLatest()。许多其他 Rx 参数是无用的，因为它们需要 onComplete() 调用，而这里我处理的是无限 Observables。

Answer 2

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

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

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) {
            }
        });
    });
}

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

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

            });

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