javascript - 计数、间隔和事件的响应式扩展缓冲区

Question

我想缓冲发送到我的服务器的事件。刷新缓冲区的触发器是已达到缓冲区大小、已达到缓冲期或已卸载窗口。

我通过创建主题并使用带有关闭通知程序的 buffer 来缓冲发送到我的服务器的事件。我使用 race 作为关闭通知程序，并使用 window.beforeunload 事件来竞争缓冲期。

this.event$ = new Subject();
this.bufferedEvent$ = this.event$
    .buffer(
        Observable.race(
            Observable.interval(bufferPeriodMs),
            Observable.fromEvent(window, 'beforeunload')
        )
    )
    .filter(events => events.length > 0)
    .switchMap(events =>
        ajax.post(
            this.baseUrl + RESOURCE_URL,
            {
                entries: events,
            },
            {
                'Content-Type': 'application/json',
            }
       )
    );

问题是，我现在如何同时限制缓冲区的大小。即，当缓冲区有 10 个项目时，我从不希望它被刷新。

Answer 1

这是我的工作解决方案。添加了额外的 console.log() 以显示事件的顺序。

唯一有点麻烦的是fullBufferTrigger中的.skip(1)，但是需要它，因为它会在缓冲区满时触发(natch)，但是中的缓冲区bufferedEvent$ 在触发之前似乎没有最新的事件。

幸运的是，有了 timeoutTrigger，最后一个事件被触发了。没有超时，fullBufferTrigger 本身不会发出最终事件。

此外，将 buffer 更改为 bufferWhen，因为前者似乎没有通过两个触发器触发，尽管您希望它来自文档。
footnote 与 buffer(race()) 比赛只完成一次，因此无论哪个触发器首先到达那里都将被使用，其他触发器将被忽略。相比之下，bufferWhen(x => race()) 会在每次事件发生时进行评估。

const bufferPeriodMs = 1000

const event$ = new Subject()
event$.subscribe(event => console.log('event$ emit', event))

// Define triggers here for testing individually
const beforeunloadTrigger = Observable.fromEvent(window, 'beforeunload')
const fullBufferTrigger = event$.skip(1).bufferCount(2)
const timeoutTrigger = Observable.interval(bufferPeriodMs).take(10)

const bufferedEvent$ = event$
  .bufferWhen( x => 
    Observable.race(
      fullBufferTrigger,
      timeoutTrigger
    )
  )
  .filter(events => events.length > 0)

// output
fullBufferTrigger.subscribe(x => console.log('fullBufferTrigger', x))
timeoutTrigger.subscribe(x => console.log('timeoutTrigger', x))
bufferedEvent$.subscribe(events => {
  console.log('subscription', events)
})

// Test sequence
const delayBy = n => (bufferPeriodMs * n) + 500 
event$.next('event1')
event$.next('event2')
event$.next('event3')

setTimeout( () => {
  event$.next('event4')
}, delayBy(1))

setTimeout( () => {
  event$.next('event5')
}, delayBy(2))

setTimeout( () => {
  event$.next('event6')
  event$.next('event7')
}, delayBy(3))

工作示例:CodePen

编辑:触发缓冲区的替代方法

由于 bufferWhen 和 race 的组合可能有点低效(比赛在每次事件发射时重新开始)，另一种方法是将触发器合并到一个流中，然后使用一个简单的缓冲区

const bufferTrigger$ = timeoutTrigger
  .merge(fullBufferTrigger)
  .merge(beforeunloadTrigger)

const bufferedEvent$ = event$
  .buffer(bufferTrigger$)
  .filter(events => events.length > 0)