You can create custom stream logic by subclassing Minipass and overriding write(chunk, encoding, callback) and end(chunk, encoding, callback). This is useful for creating logging wrappers, custom encoders, or decoders.
Example: NDJSON Encoder
To create a transform that converts objects into newline-delimited JSON:
class NDJSONEncode extends Minipass {
write(obj, cb) {
try {
return super.write(JSON.stringify(obj) + '\n', 'utf8', cb)
} catch (er) {
this.emit('error', er)
}
}
end(obj, cb) {
if (typeof obj === 'function') {
cb = obj
obj = undefined
}
if (obj !== undefined) {
this.write(obj)
}
return super.end(cb)
}
}
Example: NDJSON Decoder
To create a transform that parses newline-delimited JSON into objects (requires objectMode: true):
class NDJSONDecode extends Minipass {
constructor(options) {
super({ objectMode: true })
this._jsonBuffer = ''
}
write(chunk, encoding, cb) {
if (typeof chunk === 'string' && typeof encoding === 'string' && encoding !== 'utf8') {
chunk = Buffer.from(chunk, encoding).toString()
} else if (Buffer.isBuffer(chunk)) {
chunk = chunk.toString()
}
if (typeof encoding === 'function') {
cb = encoding
}
const jsonData = (this._jsonBuffer + chunk).split('\n')
this._jsonBuffer = jsonData.pop()
for (let i = 0; i < jsonData.length; i++) {
try {
super.write(JSON.parse(jsonData[i]))
} catch (er) {
this.emit('error', er)
continue
}
}
if (cb) cb()
}
}
