javascript - 使用 RxJS 轮询可以恢复丢失的事件

Question

我正在尝试使用 RxJS 来轮询事件。但是，我只能访问一个函数，即 getEvent()。我可以用 getEvent 函数做两件事:

1. getEvent("latest") — 这会给我最新的事件对象
2. getEvent(eventId) - 我传入一个整数，它会给我对应于 eventId 的事件对象。

事件 ID 总是从 0 开始递增，但问题是，如果我的轮询间隔不够小，我可能会错过事件。

例如，如果我执行 getEvent("latest") 并且我得到一个 ID 为 1 的事件，那就太好了。但如果我下次调用它时，我得到的 ID 为 3，我就知道我错过了一个事件。

在这种情况下，我想使用高阶可观察对象来调用 getEvent(2) 和 getEvent(3) 以便我是流的消费者创建者不必担心错过任何事件。

现在，我只有这样的东西:

timer(0, 500).pipe(
  concatMap(() => from(getEvent("latest"))
)

对于某些上下文，我正在处理这篇博文:https://itnext.io/polling-using-rxjs-b56cd3531815

Answer 1

使用 expand 递归调用 GET 非常适合这里。这是 an example with DEMO :

const source = timer(0, 2000)

const _stream = new Subject();
const stream = _stream.asObservable();

const s1 = source.pipe(tap(random)).subscribe()    

const sub = stream.pipe(
  startWith(0),
  pairwise(),  
  concatMap((v: Array<number>) => {
    let missing = v[1] - v[0];
    return missing ? getMissing(v[0], missing) : EMPTY
  })
).subscribe(console.log)

function getMissing(start, count) {
  return getById(start).pipe(
    expand(id => getById(id+1)),
    take(count)
  )  
}

// helper functions for DEMO

let i = 1;
function random() {. // THIS IS YOUR getEvent('latest')
  if (i < 10) {
    i+=2;
    _stream.next(i
      // (Math.floor(Math.random() * 8))
    )
  }
}

function getById(id) {.  // THIS IS YOUR getEvent(eventId)
  return of(id).pipe(delay(1000)) // delay to mimic network
}