Added the ability to build a transport using a transform

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 `fn` returns a [`Readable`](https://nodejs.org/api/stream.html#stream_class_stream_readable), we will 
+wrap that readable and the split2 instance 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))
+      }
+    })
+  })
+}
+
+function buildDestination () {
+  return build(function (source) {
+    source.on('data', function (obj) {
+      console.log(obj)
+    })
+  })
+}
+
+pipeline(buildTransform(), buildDestination(), function (err) {
+  console.log('pipeline completed!, err)
+})
+```
+
 ## License
 
 MIT

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,8 @@ module.exports = function build (fn, opts = {}) {
 
     // set it to null to not retain a reference to the promise
     res = null
+  } else if (res && typeof res.read === 'function') {
+    return duplexify(stream, res)
   }
 
   return stream

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": {

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
+  })
+
+  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()
+})