javascript - 这个 rxjs 合并逻辑可以简化吗？

Question

我有一个来自 Firebase 客户端的可观察对象 (onAuthStateChanged):

1. 如果用户未登录，则立即发出null，并且
2. 发出 null，如果用户 登录，片刻之后发出 用户对象。

const observable = new Observable((obs) => {
    return app.auth().onAuthStateChanged(
        obs.next,
        obs.error,
        obs.complete
    )
})

我想要的是:

1. 在应用生命周期的前 1000 毫秒内忽略任何发出的 null 值(接受 1000 毫秒后出现的 null)
2. 无论什么时候总是发出用户对象
3. 如果前 1000 毫秒内没有用户对象，则在第 1000 毫秒标记时发出 null

这是我所做的(而且似乎有效)。但是，我不愿意使用这段代码，因为它看起来不那么简洁:

const o1 = observable.pipe(skipUntil(timer(1000)))

const o2 = observable.pipe(
    takeUntil(o1),
    filter((user) => user !== null)
)

const o3 = timer(1000).pipe(takeUntil(o2), mapTo(null))

merge(o1, o2, o3).subscribe({
    next: setUser,
    error: console.log,
    complete: () => console.log("error: obs completed, this shouldn't happen"),
})

有没有不合并的方法？我试着浏览文档，但我完全迷路了。

感谢您的帮助!

Answer 1

你可以使用 concat而不是合并。将其视为使用第一个源直到完成，然后使用第二个源。

const nonNullUser = firebaseUser.pipe(
    filter(user => user !== null),
    takeUntil(timer(1000))
);

const user = concat(nonNullUser, firebaseUser);

user.subscribe(...);

---

我刚刚意识到此解决方案不会显式执行步骤 #3“在 1000 毫秒标记处发出 null”。我在想订阅 firebaseUser 会发出最新的值。但是，我不确定这是否适用于您的场景。

如果没有，我们可以通过添加 shareReplay 轻松实现。像这样:

const firebaseUser = observable.pipe(shareReplay(1));