Shouty Stream

1-before.js

const { Transform } = require('stream')
const { createGzip } = require('zlib')
const fs = require('fs')

const testfile = fs.createWriteStream('test.gz')

const createShoutyStream = () => new Transform({
  construct(cb) {
    this.data = ''
    cb()
  },
  transform(chunk, enc, cb) {
    this.data += chunk
    cb()
  },
  flush(cb) {
    this.push(this.data.toUpperCase())
    this.push(null)
  }
})

const shout1 = createShoutyStream()
const shout2 = createShoutyStream()

shout1.pipe(createGzip()).pipe(testfile, { end: false })
shout2.pipe(createGzip()).pipe(testfile, { end: false })

shout1.end('abc')
shout2.end('xyz')

2-after.js

const { Transform, PassThrough, pipeline } = require('stream')
const { createGzip } = require('zlib')
const fs = require('fs')

const testfile = fs.createWriteStream('test.gz')

const createShoutyStream = () => new Transform({
  construct(cb) {
    this.data = ''
    cb()
  },
  transform(chunk, enc, cb) {
    this.data += chunk
    cb()
  },
  flush(cb) {
    this.push(this.data.toUpperCase())
    this.push(null)
  }
})

const shout1 = createShoutyStream()
const shout2 = createShoutyStream()
const concat1 = new PassThrough()
const concat2 = new PassThrough()
concat1.cork()
concat2.cork()

const plcb = err => {
  if (err) {
    console.error('Pipeline failed:', err)
  } else {
    console.log('Pipeline success.')
  }
}

pipeline(shout1, createGzip(), concat1, plcb).pipe(testfile, { end: false })
pipeline(shout2, createGzip(), concat2, plcb).pipe(testfile, { end: false })

shout1.end('abc')
shout2.end('xyz')