Repository: pinojs/pino
Pull request #1003 (2021-04-07 10:10:25)

pino.transport()
This add  `pino.transport()` based on [`thread-stream`](http://npm.im/thread-stream).

In theory, this supersedes #937 as we internalize the pinole part.

__Supports Node v12.20.0 onwards__.

Things that are missing:

* [ ] Docs
* [x] transport example/boilerplate
* [x] Support for transport packages
* ~~Support for pure function transports (is this actually needed?)~~
* [x] More tests
* [ ] pino-specific benchmarks
* [x] test support for transport packages on windows


jsumners: > * Support for pure function transports (is this actually needed?)

I think so. 
mcollina: > > * Support for pure function transports (is this actually needed?)

> I think so. 

Why is that so? I'm not really understanding the need of it.
Processing one log line at a time is far from fast and it will generate a significant build-up in memory.
jsumners: > Processing one log line at a time is far from fast

I suppose I didn't quite follow what was meant by the item.
mcollina: pinole expects a transport to be a pure function: https://github.com/pinojs/pinole/blob/f347c04d4a6c61e05e6ac821e510548aab928458/lib/worker.mjs#L18-L28.
Then it uses `await` to process every incoming chunk of data one at a time.
mcollina: @jsumners @davidmarkclements I've just pushed some WIP code on how to create a multitransport worker. I'm not sure that the API we exposed to the end user is good - feedbacks are welcome.

mcollina: @jsumners @davidmarkclements this is almost done. Everything works when sending the data to pino-elasticsearch with https://github.com/pinojs/pino-elasticsearch/pull/82.
jsumners: > @jsumners @davidmarkclements this is almost done. Everything works when sending the data to pino-elasticsearch with [pinojs/pino-elasticsearch#82](https://github.com/pinojs/pino-elasticsearch/pull/82).

Exellent. I'm going to put this in draft mode so I can see when it is truly ready for re-review.
davidmarkclements: I think we drop the "Support for pure function transports (is this actually needed?)" and do that separately (if at all)
davidmarkclements: @mcollina I've modified the API slightly, let me know what you think (https://github.com/pinojs/pino/pull/1003/commits/21abdac531a83ab64bf1d86d28f1df8a2f34f041)

`pino.transport(name, opts)` -  this is a more normative API and it feels cleaner to document. I get the workerData, workerOpts param trade-offs but I think that we should have `opts.workerData` for specific workerData and special case `opts.destination` to create a convention for transports that matches multitransport destination (either that or rename multitransport `t.destination`  to `t.dest`)


I've also added a test for error handling, such as it is. I think we need to propagate the error to the main thread (in thread stream, issue here https://github.com/pinojs/thread-stream/issues/20) and then update that error handling test (and fixture)



jsumners: > `pino.transport(name, opts)`

@davidmarkclements @mcollina would you guys be opposed to making the signature `pino.transport({ name, opts })`? It'll give us named parameters and flexibility for expansion in the future and avoid the mystery/complexity of positional parameters.

davidmarkclements: @jsumners good suggestion, I have no issue with it. @mcollina  ?
mcollina: > @jsumners good suggestion, I have no issue with it. @mcollina ?

I have none.

However how do you implement the multitransport case?

```js
pino.transport({
  workerData: { hello: 'world' },
  destinations: [{ 
    level: 'warn',  
    destination: file
  }, { 
    level: 'info',  
    module: '/Users/matteo/repositories/pino-elasticsearch/lib.js',  
    opts: {    node: 'http://localhost:9200'  }
  }, {  
    level: 'info',  
    prettyPrint: true
  }]
})
jsumners: > However how do you implement the [multitransport case](https://github.com/pinojs/pino/pull/1003/files#diff-9939b0546354fb8ac6901a43bfc6d8d8e5fc161c00f4b89733dc3edcaf2ab2e9R49-R51)?

Looking at the code, my suggestion will _already_ remove complexity introduced by positional parameters. The `.transport()` code is already having to inspect each parameter to determine its shape and what purpose each serves. Make it a singular object parameter and we get:

```js
function transport ({
  transportScript: 'file://...',
  workerData: { hello: 'world' },
  destinations: [],
}) {}

function multitransport (input) {
  if (Array.isArray(input)) {
    return transport({
      transportScript: './worker.js',
      transports: input // or whatever
    })
  }
  
  return transport(input)
}
```
mcollina: I've pushed a commit that fixes support for v12. I'm going to work on the API now.
mcollina: I've refactored the API. Let me know what you think!
jsumners: > I've refactored the API. Let me know what you think!

Looks good. Just one question about `Promise.all` usage.
mcollina: @davidmarkclements could you take a look? I think we have made some good progress so far.
davidmarkclements: The API looks really cool, I think there's a low hanging fruit here as well:

If we

* allow options to also be set via environment variables
* allow a stream to be the default export

We can support something like

```js
const myStream = await import('my-stream') // point is  transports we make use of TLA
const { PINO_TRANSPORT_OPTS } = process.env
export default myStream(JSON.parse(PINO_TRANSPORT_OPTS))
```

This can perhaps be a separate PR though. 
davidmarkclements: When you have an options API that has both `source` and `destination` it seems like they're connected - you moving something from the source to the destination. Our API is saying load the source CODE execute it and then pass this `destination` option value into the exported function of that code. I think it's confusing to call it `source`. 

Why can't `module` also allow for a file path?

* `target`




mcollina: The problem with using the same option name for both files and packages was some code that I did not like very much and it looked too brittle. I would love to use better names for these.
mcollina: > The API looks really cool, I think there's a low hanging fruit here as well:
> 
> 
> 
> If we
> 
> 
> 
> * allow options to also be set via environment variables
> 
> * allow a stream to be the default export
> 
> 
> 
> We can support something like
> 
> 
> 
> ```js
> 
> const myStream = await import('my-stream') // point is  transports can make use of TLA
> 
> const { PINO_TRANSPORT_OPTS } = process.env
> 
> export default myStream(JSON.parse(PINO_TRANSPORT_OPTS))
> 
> ```
> 
> 
> 
> This can perhaps be a separate PR though.

I don't think this is feasible because the transports files are not top level.


davidmarkclements: I'm not sure about the implicit API for pretty printing and file writing via the `destinations` option

I think these should be referred to as built transports and be declared similarly to external or path based transports:

```js
const transport = pino.transport([{
  level: 'warn',
  module: '#pino/writer',
  options: { destination: file }
},  {
  level: 'info',
  module: '#pino/pretty',
  prettyPrint: true
}])
```

I think maybe `module` should be called `target` (and we should get rid of `source`):

```js
const transport = pino.transport([{
  level: 'warn',
  target: '#pino/writer',
  options: { destination: file }
}, 
{
  level: 'trace',
  target: './path/to/local/file.mjs',
  options: {
     consistent: true
  }
},
{
  level: 'info',
  target: 'pino-elasticsearch',
  options: {
    node: 'http://localhost:9200'
  }
},
 {
  level: 'info',
  target: '#pino/pretty',
  prettyPrint: true
}])
```



davidmarkclements: > I don't think this is feasible because the transports files are not top level.

That doesn't stop TLA - if the files are `await import`ed  it shouldn't make any difference?
davidmarkclements: > The problem with using the same option name for both files and packages was some code that I did not like very much and it looked too brittle. I would love to use better names for these.

Suggesting `target` for all cases:

* if `target` begins with `.` or `/` it's a path (or `file://`)
* if `target` begins with `#` its a built-in
* otherwise `target` is a module
mcollina: > > I don't think this is feasible because the transports files are not top level.
> 
> 
> 
> That doesn't stop TLA - if the files are `await import`ed  it shouldn't make any difference?

Oh you are right! I've learned something new today.
davidmarkclements: Should `level` and built-in transport also be available for the single transport case?

Can we get of `destinations` and instead accept either an options object or an array of options objects?
davidmarkclements: Taken a first pass at docs (written to the suggestions made rather than current state). Feedback welcome :D 
mcollina: I pushed target and targets rename
mcollina: @jsumners I think this is ready to go. Could you take a look so we can ship it?
kibertoad: @mcollina please ping me to revise types after this lands
mcollina: I've addressed most comments @jsumners 

Patch:
diff --git a/README.md b/README.md
index 95e0852..6afcbee 100644
--- a/README.md
+++ b/README.md
@@ -75,9 +75,10 @@ format logs during development:
 
 Due to Node's single-threaded event-loop, it's highly recommended that sending,
 alert triggering, reformatting and all forms of log processing
-is conducted in a separate process. In Pino parlance we call all log processors
-"transports", and recommend that the transports be run as separate
-processes, piping the stdout of the application to the stdin of the transport.
+be conducted in a separate process or thread. 
+
+In Pino terminology we call all log processors "transports", and recommend that the
+transports be run in a worker thread using our `pino.transport` API.
 
 For more details see our [Transports⇗](docs/transports.md) document.
 
diff --git a/docs/api.md b/docs/api.md
index 03c1c48..6b5872d 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -23,6 +23,7 @@
   * [logger.version](#version)
 * [Statics](#statics)
   * [pino.destination()](#pino-destination)
+  * [pino.transport()](#pino-transport)
   * [pino.final()](#pino-final)
   * [pino.multistream()](#pino-multistream)
   * [pino.stdSerializers](#pino-stdserializers)
@@ -905,6 +906,98 @@ A `pino.destination` instance can also be used to reopen closed files
 * See [Reopening log files](/docs/help.md#reopening)
 * See [Asynchronous Logging ⇗](/docs/asynchronous.md)
 
+<a id="pino-transport"></a>
+### `pino.transport(options) => ThreadStream`
+
+Create a a stream that routes logs to a worker thread that 
+wraps around a [Pino Transport](/docs/transports.md).
+
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  target: 'some-transport',
+  options: { some: 'options for', the: 'transport' }
+})
+pino(transport)
+```
+
+Multiple transports may also be defined, and specific levels can be logged to each transport:
+
+```js
+const pino = require('pino')
+const transports = pino.transport([
+  {
+    level: 'info',
+    target: 'some-transport',
+    options: { some: 'options for', the: 'transport' }
+  },
+  {
+    level: 'trace',
+    target: '#pino/file',
+    options: { destination: '/path/to/store/logs' }
+  }
+])
+pino(transports)
+```
+
+
+For more on transports, how they work, and how to create them see the [`Transports documentation`](/docs/transports.md).
+
+* See [`Transports`](/docs/transports.md)
+* See [`thread-stream` ⇗](https://github.com/mcollina/thread-stream)
+
+#### Options
+
+* `target`:  The transport to pass logs through. This may be an installed module name, an absolute path or a built-in transport (see [Transport Builtins](#transport-builtins))
+* `options`:  An options object which is serialized (see [Structured Clone Algorithm][https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm]), passed to the worker thread, parsed and then passed to the exported transport function.
+* `worker`: [Worker thread](https://nodejs.org/api/worker_threads.html#worker_threads_new_worker_filename_options) configuration options. Additionally, the `worker` option supports `worker.autoEnd`. If this is set to `false` logs will not be flushed on process exit. It is then up to the developer to call `transport.end()` to flush logs.
+* `targets`: May be specified instead of `target`. Must be an array of transport configurations. Transport configurations include the aforementioned `options` and `target` options plus a `level` option which will send only logs above a specified level to a transport.
+
+#### Transport Builtins
+
+The `target` option may be an absolute path, a module name or builtin. 
+
+A transport builtin takes the form `#pino/<name>`.
+
+The following transport builtins are supported:
+
+#####  `#pino/file`
+
+The `#pino/file` builtin routes logs to a file (or file descriptor).
+
+The `options.destination` property may be set to specify the desired file destination.
+
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  target: '#pino/file',
+  options: { destination: '/path/to/file' }
+})
+pino(transport)
+```
+
+The `options.destination` property may also be a number to represent a file descriptor. Typically this would be `1` to write to STDOUT or `2` to write to STDERR. If `options.destination` is not set, it defaults to `1` which means logs will be written to STDOUT.
+
+The difference between using the `#pino/file` transport builtin and using `pino.destination` is that `pino.destination` runs in the main thread, whereas `#pino/file` sets up `pino.destination` in a worker thread.
+
+##### `#pino/pretty`
+
+The `#pino/pretty` builtin prettifies logs.
+
+
+By default the `#pino/pretty` builtin logs to STDOUT.
+
+The `options.destination` property may be set to log pretty logs to a file descriptor or file. The following would send the prettified logs to STDERR:
+
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  target: '#pino/pretty',
+  options: { destination: 2 }
+})
+pino(transport)
+```
+
 <a id="pino-final"></a>
 
 ### `pino.final(logger, [handler]) => Function | FinalLogger`
diff --git a/docs/transports.md b/docs/transports.md
index 13ca52c..54ddf42 100644
--- a/docs/transports.md
+++ b/docs/transports.md
@@ -1,79 +1,163 @@
 # Transports
 
-A "transport" for Pino is a supplementary tool which consumes Pino logs.
+Pino transports can be used for both transmitting and transforming log output.
 
-Consider the following example:
+The way Pino generates logs:
 
-```js
-const split = require('split2')
-const pump = require('pump')
-const through = require('through2')
+1. Reduces the impact of logging on an application to the absolute minimum.
+2. Gives greater flexibility in how logs are processed and stored.
+
+It is recommended that any log transformation or transmission is performed either
+in a seperate thread or a seperate process. 
+
+Prior to Pino v7 transports would ideally operate in a seperate process - these are
+now referred to as [Legacy Transports](#legacy-transports).
+
+From Pino v7 and upwards transports can also operate inside a [Worker Thread][worker-thread],
+and can be used or configured via the options object passed to `pino` on initialization.
 
-const myTransport = through.obj(function (chunk, enc, cb) {
-  // do the necessary
-  console.log(chunk)
-  cb()
-})
 
-pump(process.stdin, split(JSON.parse), myTransport)
+[worker-thread]: https://nodejs.org/dist/latest-v14.x/docs/api/worker_threads.html
+
+## v7+ Transports
+
+A transport is a module that exports a default function which returns a writable stream:
+
+```js
+import { Writable } from 'stream'
+export default (options) => {
+  const myTransportStream = new Writable({
+    write (chunk, enc, cb) {
+    // apply a transform and send to stdout
+    console.log(chunk.toString().toUpperCase())
+    cb()
+    }
+  })
+  return myTransportStream
+}
 ```
 
-The above defines our "transport" as the file `my-transport-process.js`.
+Let's imagine the above defines our "transport" as the file `my-transport.mjs` 
+(ESM files are supported even if the project is written in CJS).
 
-Logs can now be consumed using shell piping:
+We would set up our transport by creating a transport stream with `pino.transport`
+and passing it to the `pino` function:
 
-```sh
-node my-app-which-logs-stuff-to-stdout.js | node my-transport-process.js
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  target: '/absolute/path/to/my-transport.mjs'
+})
+pino(transport)
 ```
 
-Ideally, a transport should consume logs in a separate process to the application,
-Using transports in the same process causes unnecessary load and slows down
-Node's single threaded event loop.
+The transport code will be executed in a separate worker thread. The main thread
+will write logs to the worker thread, which will write them to the stream returned
+from the function exported from the transport file/module.
 
-## In-process transports
+The exported function can also be async. Imagine the following transport:
+
+```js
+import fs from 'fs'
+import { once } from('events')
+export default async (options) => {
+  const stream = fs.createWriteStream(opts.destination)
+  await once(stream, 'open')
+  return stream
+}
+```
 
-> **Pino *does not* natively support in-process transports.**
+While initializing the stream we're able to use `await` to perform asynchronous operations. In this 
+case waiting for the write streams `open` event. 
 
-Pino does not support in-process transports because Node processes are
-single threaded processes (ignoring some technical details). Given this
-restriction, one of the methods Pino employs to achieve its speed is to
-purposefully offload the handling of logs, and their ultimate destination, to
-external processes so that the threading capabilities of the OS can be
-used (or other CPUs).
+Let's imagine the above was published to npm with the module name `some-file-transport`.
 
-One consequence of this methodology is that "error" logs do not get written to
-`stderr`. However, since Pino logs are in a parsable format, it is possible to
-use tools like [pino-tee][pino-tee] or [jq][jq] to work with the logs. For
-example, to view only logs marked as "error" logs:
+The `options.destination` value can be set when the creating the transport stream with `pino.transport` like so:
 
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  target: 'some-file-transport',
+  options: { destination: '/dev/null' }
+})
+pino(transport)
 ```
-$ node an-app.js | jq 'select(.level == 50)'
+
+Note here we've specified a module by package rather than by relative path. The options object we provide
+is serialized and injected into the transport worker thread, then passed to the module's exported function.
+This means that the options object can only contain types that are supported by the 
+[Structured Clone Algorithm][sca] which is used to (de)serializing objects between threads.
+
+What if we wanted to use both transports, but send only error logs to `some-file-transport` while
+sending all logs to `my-transport.mjs`? We can use the `pino.transport` function's `destinations` option:
+
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  destinations: [
+    { target: '/absolute/path/to/my-transport.mjs', level: 'error' },
+    { target: 'some-file-transport', options: { destination: '/dev/null' }
+  ]
+})
+pino(transport)
 ```
 
-In short, the way Pino generates logs:
+For more details on `pino.transport` see the [API docs for `pino.transport`][pino-transport].
 
-1. Reduces the impact of logging on an application to the absolute minimum.
-2. Gives greater flexibility in how logs are processed and stored.
 
-Given all of the above, Pino recommends out-of-process log processing.
+[pino-transport]: /docs/api.md#pino-transport
+[sca]: https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm
+
+
+
+## Legacy Transports
+
+A legacy Pino "transport" is a supplementary tool which consumes Pino logs.
+
+Consider the following example for creating a transport:
+
+```js
+const { pipeline, Writable } = require('stream')
+const split = require('split2')
+
+const myTransportStream = new Writable({
+  write (chunk, enc, cb) {
+  // apply a transform and send to stdout
+  console.log(chunk.toString().toUpperCase())
+  cb()
+  }
+})
+
+pipeline(process.stdin, split(JSON.parse), myTransportStream)
+```
+
+The above defines our "transport" as the file `my-transport-process.js`.
+
+Logs can now be consumed using shell piping:
 
-However, it is possible to wrap Pino and perform processing in-process.
-For an example of this, see [pino-multi-stream][pinoms].
+```sh
+node my-app-which-logs-stuff-to-stdout.js | node my-transport-process.js
+```
 
-[pino-tee]: https://npm.im/pino-tee
-[jq]: https://stedolan.github.io/jq/
-[pinoms]: https://npm.im/pino-multi-stream
+Ideally, a transport should consume logs in a separate process to the application,
+Using transports in the same process causes unnecessary load and slows down
+Node's single threaded event loop.
 
 ## Known Transports
 
 PR's to this document are welcome for any new transports!
 
+### Pino v7+ Compatible
+
++ [pino-elasticsearch](#pino-elasticsearch)
+
+### Legacy
+
 + [pino-applicationinsights](#pino-applicationinsights)
 + [pino-azuretable](#pino-azuretable)
 + [pino-cloudwatch](#pino-cloudwatch)
 + [pino-couch](#pino-couch)
 + [pino-datadog](#pino-datadog)
-+ [pino-elasticsearch](#pino-elasticsearch)
 + [pino-gelf](#pino-gelf)
 + [pino-http-send](#pino-http-send)
 + [pino-kafka](#pino-kafka)
diff --git a/example.js b/examples/basic.js
similarity index 96%
rename from example.js
rename to examples/basic.js
index 866338e..246ce7f 100644
--- a/example.js
+++ b/examples/basic.js
@@ -1,6 +1,6 @@
 'use strict'
 
-const pino = require('./')()
+const pino = require('..')()
 
 pino.info('hello world')
 pino.error('this is at error level')
diff --git a/examples/transport.js b/examples/transport.js
new file mode 100644
index 0000000..25717e4
--- /dev/null
+++ b/examples/transport.js
@@ -0,0 +1,66 @@
+'use strict'
+
+const pino = require('..')
+const { tmpdir } = require('os')
+const { join } = require('path')
+
+const file = join(tmpdir(), `pino-${process.pid}-example`)
+
+const transport = pino.transport({
+  targets: [{
+    level: 'warn',
+    target: '#pino/file',
+    options: {
+      destination: file
+    }
+  }, {
+    level: 'info',
+    target: 'pino-elasticsearch',
+    options: {
+      node: 'http://localhost:9200'
+    }
+  }, {
+    level: 'info',
+    target: '#pino/pretty'
+  }]
+})
+
+const logger = pino(transport)
+
+logger.info({
+  file
+}, 'logging destination')
+
+logger.info('hello world')
+logger.error('this is at error level')
+logger.info('the answer is %d', 42)
+logger.info({ obj: 42 }, 'hello world')
+logger.info({ obj: 42, b: 2 }, 'hello world')
+logger.info({ nested: { obj: 42 } }, 'nested')
+logger.warn('WARNING!')
+setImmediate(() => {
+  logger.info('after setImmediate')
+})
+logger.error(new Error('an error'))
+
+const child = logger.child({ a: 'property' })
+child.info('hello child!')
+
+const childsChild = child.child({ another: 'property' })
+childsChild.info('hello baby..')
+
+logger.debug('this should be mute')
+
+logger.level = 'trace'
+
+logger.debug('this is a debug statement')
+
+logger.child({ another: 'property' }).debug('this is a debug statement via child')
+logger.trace('this is a trace statement')
+
+logger.debug('this is a "debug" statement with "')
+
+logger.info(new Error('kaboom'))
+logger.info(null)
+
+logger.info(new Error('kaboom'), 'with', 'a', 'message')
diff --git a/lib/file-target.js b/lib/file-target.js
new file mode 100644
index 0000000..dde0fae
--- /dev/null
+++ b/lib/file-target.js
@@ -0,0 +1,20 @@
+'use strict'
+
+const { Transform, pipeline } = require('stream')
+const pino = require('../pino')
+const { once } = require('events')
+
+module.exports = async function (opts = {}) {
+  const stream = new Transform({
+    objectMode: true,
+    autoDestroy: true,
+    transform (chunk, enc, cb) {
+      cb(null, chunk.toString())
+    }
+  })
+
+  const destination = pino.destination({ dest: opts.destination || 1, sync: false })
+  await once(destination, 'ready')
+  pipeline(stream, destination, () => {})
+  return stream
+}
diff --git a/lib/multistream.js b/lib/multistream.js
index 3756523..69819b0 100644
--- a/lib/multistream.js
+++ b/lib/multistream.js
@@ -21,6 +21,7 @@ function multistream (streamsArray, opts) {
     write,
     add,
     flushSync,
+    end,
     minLevel: 0,
     streams: [],
     clone,
@@ -103,6 +104,15 @@ function multistream (streamsArray, opts) {
     return res
   }
 
+  function end () {
+    for (const { stream } of this.streams) {
+      if (typeof stream.flushSync === 'function') {
+        stream.flushSync()
+      }
+      stream.end()
+    }
+  }
+
   function clone (level) {
     const streams = new Array(this.streams.length)
 
diff --git a/lib/pretty-target.js b/lib/pretty-target.js
new file mode 100644
index 0000000..5a8d082
--- /dev/null
+++ b/lib/pretty-target.js
@@ -0,0 +1,29 @@
+'use strict'
+
+const pinoPretty = require('pino-pretty')
+const { Transform, pipeline } = require('stream')
+const pino = require('../pino')
+const { once } = require('events')
+
+module.exports = async function (opts = {}) {
+  const pretty = pinoPretty(opts)
+  const stream = new Transform({
+    objectMode: true,
+    autoDestroy: true,
+    transform (chunk, enc, cb) {
+      const line = pretty(chunk.toString())
+      cb(null, line)
+    }
+  })
+
+  const destination = pino.destination({ dest: opts.destination || 1, sync: false })
+  if (destination.fd === 1) {
+    // We cannot close the output
+    destination.end = function () {
+      this.emit('close')
+    }
+  }
+  await once(destination, 'ready')
+  pipeline(stream, destination, () => {})
+  return stream
+}
diff --git a/lib/tools.js b/lib/tools.js
index d241037..1bebeed 100644
--- a/lib/tools.js
+++ b/lib/tools.js
@@ -269,7 +269,10 @@ function prettifierMetaWrapper (pretty, dest, opts) {
 
       let time = this.lastTime
 
-      if (time.match(/^\d+/)) {
+      /* istanbul ignore next */
+      if (typeof time === 'number') {
+        // do nothing!
+      } else if (time.match(/^\d+/)) {
         time = parseInt(time)
       } else {
         time = time.slice(1, -1)
diff --git a/lib/transport.js b/lib/transport.js
new file mode 100644
index 0000000..d272201
--- /dev/null
+++ b/lib/transport.js
@@ -0,0 +1,80 @@
+'use strict'
+
+const { createRequire } = require('module')
+const caller = require('get-caller-file')
+const { join, isAbsolute } = require('path')
+
+const ThreadStream = require('thread-stream')
+
+function buildStream (filename, workerData, workerOpts) {
+  const stream = new ThreadStream({
+    filename,
+    workerData,
+    workerOpts,
+    sync: true // TODO should this be configurable?
+  })
+
+  stream.on('ready', function () {
+    stream.unref()
+
+    if (workerOpts.autoEnd !== false) {
+      // TODO possibly use FinalizationGroup to automatically remove
+      // this listener if the stream goes out scope.
+      process.on('exit', autoEnd)
+
+      stream.on('close', function () {
+        process.removeListener('exit', autoEnd)
+      })
+    }
+  })
+
+  return stream
+
+  function autoEnd () {
+    stream.end()
+  }
+}
+
+function transport (fullOptions) {
+  const { targets, options = {}, worker = {} } = fullOptions
+  // This function call MUST stay in the top-level function of this module
+  const callerFile = caller()
+  const callerRequire = createRequire(callerFile)
+
+  let target = fullOptions.target
+
+  if (target && targets) {
+    throw new Error('Only one of target or targets can be specified')
+  }
+
+  if (targets) {
+    target = join(__dirname, 'worker.js')
+    options.targets = targets.map((dest) => {
+      return {
+        ...dest,
+        target: fixTarget(dest.target)
+      }
+    })
+  }
+
+  return buildStream(fixTarget(target), options, worker)
+
+  function fixTarget (origin) {
+    if (isAbsolute(origin) || origin.indexOf('file://') === 0) {
+      return origin
+    }
+
+    switch (origin) {
+      case '#pino/pretty':
+        return join(__dirname, 'pretty-target.js')
+      case '#pino/file':
+        return join(__dirname, 'file-target.js')
+      /* istanbul ignore next */
+      default:
+        // TODO we cannot test this on Windows.. might not work.
+        return callerRequire.resolve(origin)
+    }
+  }
+}
+
+module.exports = transport
diff --git a/lib/worker.js b/lib/worker.js
new file mode 100644
index 0000000..0b6c498
--- /dev/null
+++ b/lib/worker.js
@@ -0,0 +1,53 @@
+'use strict'
+
+const pino = require('../pino.js')
+const build = require('pino-abstract-transport')
+
+// This file is not checked by the code coverage tool,
+// as it is not reliable.
+
+/* istanbul ignore file */
+
+module.exports = async function ({ targets }) {
+  targets = await Promise.all(targets.map(async (t) => {
+    const toLoad = 'file://' + t.target
+    const stream = await (await import(toLoad)).default(t.options)
+    return {
+      level: t.level,
+      stream
+    }
+  }))
+  return build(process, {
+    parse: 'lines',
+    metadata: true,
+    close (err, cb) {
+      let expected = 0
+      for (const transport of targets) {
+        expected++
+        transport.stream.on('close', closeCb)
+        transport.stream.end()
+      }
+
+      function closeCb () {
+        if (--expected === 0) {
+          cb(err)
+        }
+      }
+    }
+  })
+
+  function process (stream) {
+    const multi = pino.multistream(targets)
+    // TODO manage backpressure
+    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 + '\n')
+    })
+  }
+}
diff --git a/package.json b/package.json
index 9fbde3b..20e5674 100644
--- a/package.json
+++ b/package.json
@@ -22,8 +22,8 @@
     "docs": "docsify serve",
     "browser-test": "airtap --local 8080 test/browser*test.js",
     "lint": "eslint .",
-    "test": "npm run lint && tap --100 test/*test.js test/*/*test.js && npm run test-types",
-    "test-ci": "npm run lint && tap test/*test.js test/*/*test.js --coverage-report=lcovonly && npm run test-types",
+    "test": "npm run lint && tap test/*test.js test/*/*test.js && npm run test-types",
+    "test-ci": "npm run lint && tap --no-check-coverage test/*test.js test/*/*test.js --coverage-report=lcovonly && npm run test-types",
     "test-types": "tsc && tsd",
     "cov-ui": "tap --coverage-report=html test/*test.js test/*/*test.js",
     "bench": "node benchmarks/utils/runbench all",
@@ -99,9 +99,12 @@
   "dependencies": {
     "fast-redact": "^3.0.0",
     "fast-safe-stringify": "^2.0.7",
+    "get-caller-file": "^2.0.5",
+    "pino-abstract-transport": "^0.2.0",
     "pino-std-serializers": "^4.0.0",
     "quick-format-unescaped": "^4.0.3",
-    "sonic-boom": "^2.0.1"
+    "sonic-boom": "^2.0.1",
+    "thread-stream": "^0.10.0"
   },
   "tsd": {
     "directory": "test/types"
diff --git a/pino.js b/pino.js
index 7215fdc..0ae4679 100644
--- a/pino.js
+++ b/pino.js
@@ -226,6 +226,7 @@ module.exports.destination = (dest = process.stdout.fd) => {
   }
 }
 
+module.exports.transport = require('./lib/transport')
 module.exports.multistream = require('./lib/multistream')
 
 module.exports.final = final


Test patch:
diff --git a/test/fixtures/to-file-transport.js b/test/fixtures/to-file-transport.js
new file mode 100644
index 0000000..33de462
--- /dev/null
+++ b/test/fixtures/to-file-transport.js
@@ -0,0 +1,13 @@
+'use strict'
+
+const fs = require('fs')
+const { once } = require('events')
+
+async function run (opts) {
+  if (!opts.destination) throw new Error('kaboom')
+  const stream = fs.createWriteStream(opts.destination)
+  await once(stream, 'open')
+  return stream
+}
+
+module.exports = run
diff --git a/test/fixtures/to-file-transport.mjs b/test/fixtures/to-file-transport.mjs
new file mode 100644
index 0000000..e0cbeec
--- /dev/null
+++ b/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.destination)
+  await once(stream, 'open')
+  return stream
+}
diff --git a/test/fixtures/transport/index.js b/test/fixtures/transport/index.js
new file mode 100644
index 0000000..c576a7b
--- /dev/null
+++ b/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.destination)
+  await once(stream, 'open')
+  return stream
+}
+
+module.exports = run
diff --git a/test/fixtures/transport/package.json b/test/fixtures/transport/package.json
new file mode 100644
index 0000000..26beeaa
--- /dev/null
+++ b/test/fixtures/transport/package.json
@@ -0,0 +1,5 @@
+{
+  "name": "transport",
+  "version": "0.0.1",
+  "main": "./index.js"
+}
diff --git a/test/helper.js b/test/helper.js
index dd25d9d..ed80734 100644
--- a/test/helper.js
+++ b/test/helper.js
@@ -53,7 +53,7 @@ function sleep (ms) {
 
 function watchFileCreated (filename) {
   return new Promise((resolve, reject) => {
-    const TIMEOUT = 800
+    const TIMEOUT = 2000
     const INTERVAL = 100
     const threshold = TIMEOUT / INTERVAL
     let counter = 0
@@ -72,4 +72,4 @@ function watchFileCreated (filename) {
   })
 }
 
-module.exports = { getPathToNull, sink, check, once, sleep, watchFileCreated }
+module.exports = { getPathToNull, sink, check, once, sleep, watchFileCreated, isWin }
diff --git a/test/multistream.test.js b/test/multistream.test.js
index 463be48..6bae9c8 100644
--- a/test/multistream.test.js
+++ b/test/multistream.test.js
@@ -475,3 +475,32 @@ test('flushSync', function (t) {
     })()
   })
 })
+
+test('ends all streams', function (t) {
+  t.plan(7)
+  const stream = writeStream(function (data, enc, cb) {
+    t.pass('message')
+    cb()
+  })
+  stream.flushSync = function () {
+    t.pass('flushSync')
+  }
+  // stream2 has no flushSync
+  const stream2 = writeStream(function (data, enc, cb) {
+    t.pass('message2')
+    cb()
+  })
+  const streams = [
+    { stream: stream },
+    { level: 'debug', stream: stream },
+    { level: 'trace', stream: stream2 },
+    { level: 'fatal', stream: stream },
+    { level: 'silent', stream: stream }
+  ]
+  const multi = multistream(streams)
+  const log = pino({
+    level: 'trace'
+  }, multi)
+  log.info('info stream')
+  multi.end()
+})
diff --git a/test/targets.test.js b/test/targets.test.js
new file mode 100644
index 0000000..c6bc048
--- /dev/null
+++ b/test/targets.test.js
@@ -0,0 +1,66 @@
+'use strict'
+
+const { test } = require('tap')
+const proxyquire = require('proxyquire')
+const Writable = require('stream').Writable
+
+test('pretty-target mocked', async function ({ equal, same, plan, pass }) {
+  plan(3)
+  let ret
+  const prettyTarget = proxyquire('../lib/pretty-target', {
+    '../pino': {
+      destination (opts) {
+        same(opts, { dest: 1, sync: false })
+
+        ret = new Writable()
+        ret.fd = opts.dest
+
+        process.nextTick(() => {
+          ret.emit('ready')
+        })
+
+        Object.defineProperty(ret, 'end', {
+          get () {
+            return 'unused'
+          },
+          set (end) {
+            pass('prop set')
+            const obj = {
+              emit (ev) {
+                equal(ev, 'close')
+              },
+              end
+            }
+            obj.end()
+          }
+        })
+        return ret
+      }
+    }
+  })
+
+  await prettyTarget()
+})
+
+test('file-target mocked', async function ({ equal, same, plan, pass }) {
+  plan(1)
+  let ret
+  const fileTarget = proxyquire('../lib/file-target', {
+    '../pino': {
+      destination (opts) {
+        same(opts, { dest: 1, sync: false })
+
+        ret = new Writable()
+        ret.fd = opts.dest
+
+        process.nextTick(() => {
+          ret.emit('ready')
+        })
+
+        return ret
+      }
+    }
+  })
+
+  await fileTarget()
+})
diff --git a/test/transport.test.js b/test/transport.test.js
new file mode 100644
index 0000000..4fdbcee
--- /dev/null
+++ b/test/transport.test.js
@@ -0,0 +1,368 @@
+'use strict'
+
+const os = require('os')
+const { join } = require('path')
+const { once } = require('events')
+const { readFile, symlink, unlink } = require('fs').promises
+const { test } = require('tap')
+const { isWin, watchFileCreated } = require('./helper')
+const pino = require('../')
+const url = require('url')
+const strip = require('strip-ansi')
+
+const { pid } = process
+const hostname = os.hostname()
+
+test('pino.transport with file', async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.transport({
+    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('pino.transport with file (no options + error handling)', async ({ equal }) => {
+  const transport = pino.transport({
+    target: join(__dirname, 'fixtures', 'to-file-transport.js')
+  })
+  const [err] = await once(transport, 'error')
+  equal(err.message, 'kaboom')
+})
+
+// TODO make this test pass on Windows
+test('pino.transport with package', { skip: isWin }, async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+
+  try {
+    await unlink(join(__dirname, '..', 'node_modules', 'transport'))
+  } catch {}
+
+  await symlink(
+    join(__dirname, 'fixtures', 'transport'),
+    join(__dirname, '..', 'node_modules', 'transport')
+  )
+
+  const transport = pino.transport({
+    target: 'transport',
+    options: { destination }
+  })
+  teardown(async () => {
+    await unlink(join(__dirname, '..', 'node_modules', 'transport'))
+    transport.end()
+  })
+  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('pino.transport with file URL', async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.transport({
+    target: url.pathToFileURL(join(__dirname, 'fixtures', 'to-file-transport.js')).href,
+    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('pino.transport errors if file does not exists', ({ plan, pass }) => {
+  plan(1)
+  const instance = pino.transport({
+    target: join(__dirname, 'fixtures', 'non-existent-file'),
+    worker: {
+      stdin: true,
+      stdout: true,
+      stderr: true
+    }
+  })
+  instance.on('error', function () {
+    pass('error received')
+  })
+})
+
+test('pino.transport with esm', async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.transport({
+    target: join(__dirname, 'fixtures', 'to-file-transport.mjs'),
+    options: { destination }
+  })
+  const instance = pino(transport)
+  teardown(transport.end.bind(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('pino.transport with two files', async ({ same, teardown }) => {
+  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.transport({
+    targets: [{
+      level: 'info',
+      target: join(__dirname, 'fixtures', 'to-file-transport.js'),
+      options: { destination: dest1 }
+    }, {
+      level: 'info',
+      target: join(__dirname, 'fixtures', 'to-file-transport.js'),
+      options: { destination: dest2 }
+    }]
+  })
+  teardown(transport.end.bind(transport))
+  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('pino.transport with an array including a prettyPrint destination', async ({ same, match, teardown }) => {
+  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.transport({
+    targets: [{
+      level: 'info',
+      target: '#pino/file',
+      options: {
+        destination: dest1
+      }
+    }, {
+      level: 'info',
+      target: '#pino/pretty',
+      options: {
+        destination: dest2
+      }
+    }]
+  })
+  teardown(transport.end.bind(transport))
+  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 actual = (await readFile(dest2)).toString()
+  match(strip(actual), /\[.*\] INFO.*hello/)
+})
+
+test('no transport.end()', async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.transport({
+    target: join(__dirname, 'fixtures', 'to-file-transport.js'),
+    options: { destination }
+  })
+  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('autoEnd = false', async ({ equal, same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const count = process.listenerCount('exit')
+  const transport = pino.transport({
+    target: join(__dirname, 'fixtures', 'to-file-transport.js'),
+    options: { destination },
+    worker: { autoEnd: false }
+  })
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello')
+  await watchFileCreated(destination)
+
+  equal(count, process.listenerCount('exit'))
+
+  const result = JSON.parse(await readFile(destination))
+  delete result.time
+  same(result, {
+    pid,
+    hostname,
+    level: 30,
+    msg: 'hello'
+  })
+})
+
+test('pino.transport with target and targets', async ({ fail, equal }) => {
+  try {
+    pino.transport({
+      target: '/a/file',
+      targets: [{
+        target: '/a/file'
+      }]
+    })
+    fail('must throw')
+  } catch (err) {
+    equal(err.message, 'Only one of target or targets can be specified')
+  }
+})
+
+// TODO make this test pass on Windows
+test('pino.transport with package as a target', { skip: isWin }, async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+
+  try {
+    await unlink(join(__dirname, '..', 'node_modules', 'transport'))
+  } catch {}
+
+  await symlink(
+    join(__dirname, 'fixtures', 'transport'),
+    join(__dirname, '..', 'node_modules', 'transport')
+  )
+
+  const transport = pino.transport({
+    targets: [{
+      target: 'transport',
+      options: { destination }
+    }]
+  })
+  teardown(async () => {
+    await unlink(join(__dirname, '..', 'node_modules', 'transport'))
+    transport.end()
+  })
+  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('pino.transport with target #pino/file', async ({ same, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.transport({
+    target: '#pino/file',
+    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('pino.transport with target #pino/pretty', async ({ match, teardown }) => {
+  const destination = join(
+    os.tmpdir(),
+    '_' + Math.random().toString(36).substr(2, 9)
+  )
+  const transport = pino.transport({
+    target: '#pino/pretty',
+    options: { destination }
+  })
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello')
+  await watchFileCreated(destination)
+  const actual = await readFile(destination, 'utf8')
+  match(strip(actual), /\[.*\] INFO.*hello/)
+})