javascript RxJS中的正确背压处理

Posted 2021-05-03

const { delay, tap } = require('rxjs/operators')

const createIteratorSubject = require('./createIteratorSubject')

const createGenerator = function*() {
  yield 1
  yield 2
  yield 3
  yield 4
}

const generator = createGenerator()

const iterator$ = (
  createIteratorSubject(
    generator,
  )
)

iterator$
.pipe(
  tap(value => (
    console
    .info(
      'INCOMING VALUE',
      value,
    )
  )),
  delay(1000),
  tap(value => (
    console
    .info(
      'DONE PROCESSING VALUE',
      value,
    )
  )),
)
.subscribe({
  complete: () => {
    console
    .info('DONE PROCESSING ALL VALUES')
  },
  next: (
    iterator$
    .push
  ),
})

// INCOMING VALUE 1
// DONE PROCESSING VALUE 1
// INCOMING VALUE 2
// DONE PROCESSING VALUE 2
// INCOMING VALUE 3
// DONE PROCESSING VALUE 3
// INCOMING VALUE 4
// DONE PROCESSING VALUE 4
// DONE PROCESSING ALL VALUES