javascript RxJS转换流使用

Posted 2021-05-03

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

const createTransformStreamObservable = require('./createTransformStreamObservable')

const readStream = (
  fs
  .createReadStream('./novel.txt')
)

const writeStream = (
  fs
  .createWriteStream('./uppercaseNovel.txt')
)

const transformStream$ = (
  createTransformStreamObservable(
    readStream,
  )
)

// Pipe to a `writeStream` or stream$ will never complete
transformStream$
.stream
.pipe(writeStream)

transformStream$
.pipe(
  tap(() => {
    console
    .info('INCOMING CHUNK')
  }),
  delay(1000), // This could be an AJAX call
  map(chunk => (
    chunk
    .toString()
    .toUpperCase()
  )),
  tap(() => {
    console
    .info('DONE PROCESSING CHUNK')
  }),
  tap(
    transformStream$
    .push
  ),
)
.subscribe({
  complete: () => {
    console
    .info('DONE PROCESSING ALL CHUNKS')
  },
  next: (
    transformStream$
    .next
  ),
})

// INCOMING CHUNK
// DONE PROCESSING CHUNK
// ...
// INCOMING CHUNK
// DONE PROCESSING CHUNK
// DONE PROCESSING ALL CHUNKS