Find out if the stream chunk is the last

user_735307 0
Viewed 4218

When writing NodeJS Transform stream in transform function how can I know if the chunk is the last one or there is no any new chunk coming.

_transform(chunk: any, encoding: string, callback: Function): void {
    // accumulating chunks here to buffer
    // so that I need to do some processing on the whole buffer
    // and I need to understand when to do that
}

So I need to know when the chunks coming to the Stream are over, to do some processing on the buffer composed of all the chunks and then push the processed data from the stream.

typescript node.js javascript
4 Answers

In the _transform you can not figure out if there will be more data or not.

Depending on your use case you would either listen to the end event, or use _flush:

Stream: transform._flush(callback):

Custom Transform implementations may implement the transform._flush() method. This will be called when there is no more written data to be consumed, but before the 'end' event is emitted signaling the end of the Readable stream.

Within the transform._flush() implementation, the readable.push() method may be called zero or more times, as appropriate. The callback function must be called when the flush operation is complete.

user_19604550

Using through2

fs.createReadStream('/tmp/important.dat')
  .pipe(through2(
    (chunk, enc, cb) => cb(null, chunk), // transform is a noop
    function (cb) { // FLUSH FUNCTION
      this.push('tacking on an extra buffer to the end');
      cb();
    }
  ))
  .pipe(fs.createWriteStream('/tmp/wut.txt'));
user_20876400

EDIT: WARNING DO NOT USE (I noticed that transform streams can still be pushed chunks after the source has ended or unpiped, I will see if I can come up with a solution and update this answer).

I have written a "last chunk aware transform" to demonstrate how to deal with this dilemma. Knowing at "chunk transform time" if the chunk is the last or not can be very useful (to check it afterwards on _flush() can be too late). This transform will also be compatible with streams that doesn't end it when done (unpipes instead), to try that out just un-comment the // end: false line. My code is license free, use it as you wish.

import * as stream from 'node:stream'

// for this demo
class Source extends stream.Readable {
  #offset = 0; #size =  10
  _read(preferredSize) {
    if (this.#offset > this.#size) {
      this.push(null) // end stream
    } else {
      this.push(Buffer.from([this.#offset++]))
    }
  }
}

// for this demo (to inspect result)
class Output extends stream.Writable {
  _write(chunk, _encoding, callback) {
    console.log(chunk); callback()
  }
  _final(callback) {
    console.log('end'); callback()
  }
}

// last chunk aware transform
class Transform extends stream.Transform {
  #prevChunk
  #endOnUnpipe = true

  constructor() {
    super()
    this.on('unpipe', this.#onUnpipe)
  }

  _transform(chunk, _encoding, callback) {
    if (!this.#prevChunk) {
      this.#prevChunk = chunk
      callback()
    } else {
      this.#transform(this.#prevChunk, false, callback)
      this.#prevChunk = chunk
    }
    
  }

  #transform(chunk, isLast, callback) {
    if (isLast) chunk[0] = 0xFF
    this.push(chunk) // no change
    callback()
  }

  _flush(callback) {
    console.log('flush')
    this.#transform(this.#prevChunk, true, callback)
  }

  // if source did not end our transform
  #onUnpipe() {
    if (!this.writableEnded) {
      console.log('unpipe')
      if (this.#endOnUnpipe) {
        this.end() // will trigger _flush
      } else {
        this.#transform(this.#prevChunk, true, () => {})
      }
    }
  }
}

const
  source    = new Source(),
  transform = new Transform(),
  output    = new Output()

transform.pipe(output)
source.pipe(transform, {
  // end: false // whether to end transform or not at end of source
})

Usage notes: This works fine unless the stream can suddenly pause for a long time, then the transform of the last chunk will be delayed until it ends or more data is received. So be aware of this if that could be a problem for you.

user_42161530
Related