fix(transport): throw an error in case transport last target is Readable

fix pinojs#1774

lib/worker-pipeline.js

@@ -15,6 +15,16 @@ module.exports = async function ({ targets }) {
     const stream = await fn(t.options)
     return stream
   }))
+
+  if (streams.length === 0) {
+    throw new Error('targets must not be empty')
+  }
+
+  // If not readable === is writeable
+  if (typeof streams[streams.length - 1]._read === 'function') {
+    throw new Error('last target should not be readable (Duplex and Transform are readable as well), instead it should be a Writable stream')
+  }
+
   const ee = new EE()
 
   const stream = new PassThrough({

test/transport/pipeline.test.js

@@ -34,3 +34,49 @@ test('pino.transport with a pipeline', async ({ same, teardown }) => {
     service: 'pino' // this property was added by the transform
   })
 })
+
+test('pino.transport with a pipeline and last transport is writable and not readable', async ({ same, teardown }) => {
+  const destination = file()
+  const transport = pino.transport({
+    pipeline: [{
+      target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+      options: { destination }
+    }]
+  })
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello')
+  await watchFileCreated(destination)
+  const result = JSON.parse(await readFile(destination))
+  delete result.time
+  same(result, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})
+
+test('should fail when pino.transport with a pipeline have no transports', ({ plan, equal }) => {
+  plan(1)
+  const pipelinedInstance = pino.transport({
+    pipeline: []
+  })
+
+  pipelinedInstance.on('error', function (e) {
+    equal(e.message, 'targets must not be empty')
+  })
+})
+
+test('should fail when pino.transport with a pipeline that last transport is readable', ({ plan, equal }) => {
+  plan(1)
+  const pipelinedInstance = pino.transport({
+    pipeline: [{
+      target: join(__dirname, '..', 'fixtures', 'transport-transform.js')
+    }]
+  })
+
+  pipelinedInstance.on('error', function (e) {
+    equal(e.message, 'last target should not be readable (Duplex and Transform are readable as well), instead it should be a Writable stream')
+  })
+})