javascript - RxJS 根据外部条件从流中提取数据

Question

上下文:我有一个应用程序的日志库，我想使用 Rx subject 将日志写入数据库。 问题:日志是在建立与数据库的连接之前生成的，并且该连接可能随时变得不可用，但我可以观察到它要么包含具有使用数据库的方法的对象，要么包含 undefined object 。我想要做的是在数据库持久性未定义时将日志保留在流中，并在可用时恢复。

我认为代码应该如下所示:

logsSubject
           .takeWhen(/* test for database persistence */) // made up name because I don't know a Rx method that does what I want
           .subscribe(/* write data to database */);

但是我不确定这是否真的可能，因为我使用 RxJS 的时间很短

Answer 1

以下代码用于模拟您的场景。数据库每 10 秒连接一次，连接后每 5 秒断开一次。在实际用例中，连接和断开连接应该都是共享的 Observable 或Subject。我在这里使用合并而不是 takeUntil 运算符，因为 takeUntil 将刚刚完成可观察的序列，并且您将无法恢复它。 (repeatWhen 运算符应该可以工作，但在 Rxjs 5 中似乎缺失)

//emit value every 1s
const logWrite = Rx.Observable.interval(1000)

const dbConnected = new Rx.Subject()
// make it fire every 10s 
Rx.Observable.interval(10000).do(()=>dbConnected.next('db 
connected')).subscribe()

const dbDisconnected = Rx.Observable.timer(3000)
.flatMap(()=>Rx.Observable.throw('db disconnected'))


const writeWhenConnected = logWrite.merge(dbDisconnected)
.retryWhen(function(errors) {
    return errors.switchMap(()=>dbConnected);
});

writeWhenConnected.subscribe(val => console.log(val),console.warn);