Added the ability to build a transport using a transform (#25)

* Added the ability to build it using a transform * Update README.md Co-authored-by: James Sumners <[email protected]> * Update README.md Co-authored-by: James Sumners <[email protected]> * Update README.md Co-authored-by: Manuel Spigolon <[email protected]> * Added opts.transform * transform -> enablePipelining * Update index.js Co-authored-by: Manuel Spigolon <[email protected]> Co-authored-by: James Sumners <[email protected]> Co-authored-by: Manuel Spigolon <[email protected]>
pinojs · Oct 2, 2021 · de2ca57 · de2ca57
1 parent a29136d
commit de2ca57
Show file tree

Hide file tree

Showing 4 changed files with 99 additions and 0 deletions.
diff --git a/README.md b/README.md
@@ -51,6 +51,10 @@ Create a [`split2`](http://npm.im/split2) instance and returns it.
 This same instance is also passed to the given function, which is called
 synchronously.
 
+If `opts.transform` is `true`, `pino-abstract-transform` will 
+wrap the split2 instance and the returned stream using [`duplexify`](https://www.npmjs.com/package/duplexify),
+so they can be concatenated into multiple transports.
+
 #### Events emitted
 
 In addition to all events emitted by a [`Readable`](https://nodejs.org/api/stream.html#stream_class_stream_readable)
@@ -97,6 +101,40 @@ module.exports = function (opts) {
 }
 ```
 
+### Stream concatenation / pipeline
+
+You can pipeline multiple transports:
+
+```js
+const build = require('pino-abstract-transport')
+const { Transform, pipeline } = require('stream')
+
+function buildTransform () {
+  return build(function (source) {
+    return new Transform({
+      objectMode: true,
+      autoDestroy: true,
+      transform (line, enc, cb) {
+        line.service = 'bob'
+        cb(null, JSON.stringify(line))
+      }
+    })
+  }, { enablePipelining: true })
+}
+
+function buildDestination () {
+  return build(function (source) {
+    source.on('data', function (obj) {
+      console.log(obj)
+    })
+  })
+}
+
+pipeline(process.stdin, buildTransform(), buildDestination(), function (err) {
+  console.log('pipeline completed!', err)
+})
+```
+
 ## License
 
 MIT
diff --git a/index.js b/index.js
@@ -2,6 +2,7 @@
 
 const metadata = Symbol.for('pino.metadata')
 const split = require('split2')
+const duplexify = require('duplexify')
 
 module.exports = function build (fn, opts = {}) {
   const parseLines = opts.parse === 'lines'
@@ -65,6 +66,10 @@ module.exports = function build (fn, opts = {}) {
 
     // set it to null to not retain a reference to the promise
     res = null
+  } else if (opts.enablePipelining && res) {
+    return duplexify(stream, res, {
+      objectMode: true
+    })
   }
 
   return stream

diff --git a/package.json b/package.json
@@ -23,6 +23,7 @@
   },
   "homepage": "https://github.com/pinojs/pino-abstract-transport#readme",
   "dependencies": {
+    "duplexify": "^4.1.2",
     "split2": "^3.2.2"
   },
   "devDependencies": {

diff --git a/test/base.test.js b/test/base.test.js
@@ -1,6 +1,7 @@
 'use strict'
 
 const { once } = require('events')
+const { Transform, pipeline } = require('stream')
 
 const { test } = require('tap')
 const build = require('../')
@@ -380,3 +381,57 @@ test('close with promises', ({ same, plan, pass }) => {
   stream.write(lines)
   stream.end()
 })
+
+test('support Transform streams', ({ same, plan, error }) => {
+  plan(7)
+
+  const expected1 = [{
+    level: 30,
+    time: 1617955768092,
+    pid: 2942,
+    hostname: 'MacBook-Pro.local',
+    msg: 'hello world'
+  }, {
+    level: 30,
+    time: 1617955768092,
+    pid: 2942,
+    hostname: 'MacBook-Pro.local',
+    msg: 'another message',
+    prop: 42
+  }]
+
+  const expected2 = []
+
+  const stream1 = build(function (source) {
+    const transform = new Transform({
+      objectMode: true,
+      autoDestroy: true,
+      transform (chunk, enc, cb) {
+        same(expected1.shift(), chunk)
+        chunk.service = 'from transform'
+        expected2.push(chunk)
+        cb(null, JSON.stringify(chunk) + '\n')
+      }
+    })
+
+    pipeline(source, transform, () => {})
+
+    return transform
+  }, { enablePipelining: true })
+
+  const stream2 = build(function (source) {
+    source.on('data', function (line) {
+      same(expected2.shift(), line)
+    })
+  })
+
+  pipeline(stream1, stream2, function (err) {
+    error(err)
+    same(expected1, [])
+    same(expected2, [])
+  })
+
+  const lines = expected1.map(JSON.stringify).join('\n')
+  stream1.write(lines)
+  stream1.end()
+})