pino.transport()

lib/multitransport/index.js

@@ -0,0 +1,8 @@
+'use strict'
+
+const transport = require('../transport')
+const { join } = require('path')
+
+module.exports = function multitransport (definitions, workerOpts) {
+  return transport(join(__dirname, 'worker.js'), definitions, workerOpts)
+}

lib/multitransport/worker.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const pino = require('../../pino.js')
+const build = require('pino-abstract-transport')
+
+module.exports = async function ({ transports }) {
+  transports = await Promise.all(transports.map(async (t) => {
+    const stream = await (await import(t.module)).default(t.opts)
+    return {
+      level: t.level,
+      stream
+    }
+  }))
+  return build(process, { parse: 'lines', metadata: true })
+
+  function process (stream) {
+    const multi = pino.multistream(transports)
+    stream.on('data', function (chunk) {
+      const { lastTime, lastMsg, lastObj, lastLevel } = this
+      multi.lastLevel = lastLevel
+      multi.lastTime = lastTime
+      multi.lastMsg = lastMsg
+      multi.lastObj = lastObj
+
+      // TODO handle backpressure
+      multi.write(chunk)
+    })
+
+    stream.on('end', function () {
+      for (const transport of transports) {
+        transport.stream.end()
+      }
+    })
+  }
+}

lib/transport.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const { createRequire } = require('module')
+const caller = require('get-caller-file')
+const { isAbsolute } = require('path')
+
+const ThreadStream = require('thread-stream')
+
+function transport (filename, workerData, workerOpts = {}) {
+  if (!(isAbsolute(filename) || filename.indexOf('file://') === 0)) {
+    const callerFile = caller()
+    const callerRequire = createRequire(callerFile)
+    filename = callerRequire.resolve(filename)
+  }
+
+  const stream = new ThreadStream({
+    filename,
+    workerData,
+    workerOpts,
+    sync: true // TODO should this be configurable?
+  })
+
+  stream.on('ready', function () {
+    stream.unref()
+  })
+
+  return stream
+}
+
+module.exports = transport

package.json

@@ -87,14 +87,18 @@
     "tap": "^15.0.1",
     "tape": "^5.0.0",
     "through2": "^4.0.0",
+    "transport": "./test/fixtures/transport",
     "winston": "^3.3.3"
   },
   "dependencies": {
     "fast-redact": "^3.0.0",
     "fast-safe-stringify": "^2.0.7",
     "flatstr": "^1.0.12",
+    "get-caller-file": "^2.0.5",
+    "pino-abstract-transport": "^0.1.0",
     "pino-std-serializers": "^3.1.0",
     "quick-format-unescaped": "^4.0.3",
-    "sonic-boom": "^1.0.2"
+    "sonic-boom": "^1.0.2",
+    "thread-stream": "^0.8.0"
   }
 }

pino.js

@@ -223,7 +223,9 @@ module.exports.destination = (dest = process.stdout.fd) => {
   }
 }
 
+module.exports.transport = require('./lib/transport')
 module.exports.multistream = require('./lib/multistream')
+module.exports.multitransport = require('./lib/multitransport')
 
 module.exports.final = final
 module.exports.levels = mappings()

test/fixtures/to-file-transport.js

@@ -0,0 +1,12 @@
+'use strict'
+
+const fs = require('fs')
+const { once } = require('events')
+
+async function run (opts) {
+  const stream = fs.createWriteStream(opts.dest)
+  await once(stream, 'open')
+  return stream
+}
+
+module.exports = run

test/fixtures/to-file-transport.mjs

@@ -0,0 +1,8 @@
+import { createWriteStream } from 'fs'
+import { once } from 'events'
+
+export default async function run (opts) {
+  const stream = createWriteStream(opts.dest)
+  await once(stream, 'open')
+  return stream
+}

test/fixtures/transport/index.js

@@ -0,0 +1,12 @@
+'use strict'
+
+const fs = require('fs')
+const { once } = require('events')
+
+async function run (opts) {
+  const stream = fs.createWriteStream(opts.dest)
+  await once(stream, 'open')
+  return stream
+}
+
+module.exports = run

test/fixtures/transport/package.json

@@ -0,0 +1,5 @@
+{
+  "name": "transport",
+  "version": "0.0.1",
+  "main": "./index.js"
+}

test/multitransport.test.js

@@ -0,0 +1,52 @@
+'use strict'
+
+const os = require('os')
+const { join } = require('path')
+const { readFile } = require('fs').promises
+const { test } = require('tap')
+const { watchFileCreated } = require('./helper')
+const pino = require('../')
+
+const { pid } = process
+const hostname = os.hostname()
+
+test('pino.multitransport with two files', async ({ same }) => {
+  const dest1 = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const dest2 = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.multitransport({
+    transports: [{
+      level: 'info',
+      module: join(__dirname, 'fixtures', 'to-file-transport.js'),
+      opts: { dest: dest1 }
+    }, {
+      level: 'info',
+      module: join(__dirname, 'fixtures', 'to-file-transport.js'),
+      opts: { dest: dest2 }
+    }]
+  })
+  const instance = pino(transport)
+  instance.info('hello')
+  await Promise.all([watchFileCreated(dest1), watchFileCreated(dest2)])
+  const result1 = JSON.parse(await readFile(dest1))
+  delete result1.time
+  same(result1, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+  const result2 = JSON.parse(await readFile(dest2))
+  delete result2.time
+  same(result2, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})

test/transport.test.js

@@ -0,0 +1,96 @@
+'use strict'
+
+const os = require('os')
+const { join } = require('path')
+const { readFile } = require('fs').promises
+const { test } = require('tap')
+const { watchFileCreated } = require('./helper')
+const pino = require('../')
+const url = require('url')
+
+const { pid } = process
+const hostname = os.hostname()
+
+test('pino.transport with file', async ({ same }) => {
+  const dest = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const instance = pino(pino.transport(join(__dirname, 'fixtures', 'to-file-transport.js'), { dest }))
+  instance.info('hello')
+  await watchFileCreated(dest)
+  const result = JSON.parse(await readFile(dest))
+  delete result.time
+  same(result, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})
+
+test('pino.transport with package', async ({ same }) => {
+  const dest = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const instance = pino(pino.transport('transport', { dest }))
+  instance.info('hello')
+  await watchFileCreated(dest)
+  const result = JSON.parse(await readFile(dest))
+  delete result.time
+  same(result, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})
+
+test('pino.transport with file URL', async ({ same }) => {
+  const dest = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const instance = pino(pino.transport(url.pathToFileURL(join(__dirname, 'fixtures', 'to-file-transport.js')).href, { dest }))
+  instance.info('hello')
+  await watchFileCreated(dest)
+  const result = JSON.parse(await readFile(dest))
+  delete result.time
+  same(result, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})
+
+test('pino.transport errors if file does not exists', ({ plan, pass }) => {
+  plan(1)
+  const instance = pino.transport(join(__dirname, 'fixtures', 'non-existent-file'), {}, {
+    stdin: true,
+    stdout: true,
+    stderr: true
+  })
+  instance.on('error', function () {
+    pass('error received')
+  })
+})
+
+test('pino.transport with esm', async ({ same }) => {
+  const dest = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const instance = pino(pino.transport(join(__dirname, 'fixtures', 'to-file-transport.mjs'), { dest }))
+  instance.info('hello')
+  await watchFileCreated(dest)
+  const result = JSON.parse(await readFile(dest))
+  delete result.time
+  same(result, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})