Repository: pinojs/pino
Pull request #1960 (2024-04-29 13:28:59)

build(deps): bump sonic-boom from 3.8.1 to 4.0.1
Bumps [sonic-boom](https://github.com/pinojs/sonic-boom) from 3.8.1 to 4.0.1.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/pinojs/sonic-boom/releases">sonic-boom's releases</a>.</em></p>
<blockquote>
<h2>v4.0.1</h2>
<p><strong>Full Changelog</strong>: <a href="https://github.com/pinojs/sonic-boom/compare/v4.0.0...v4.0.1">https://github.com/pinojs/sonic-boom/compare/v4.0.0...v4.0.1</a></p>
<h2>v4.0.0</h2>
<h2>What's Changed</h2>
<ul>
<li>Drop old node.js versions by <a href="https://github.com/mcollina"><code>@​mcollina</code></a> in <a href="https://redirect.github.com/pinojs/sonic-boom/pull/196">pinojs/sonic-boom#196</a></li>
</ul>
<p><strong>Full Changelog</strong>: <a href="https://github.com/pinojs/sonic-boom/compare/v3.8.1...v4.0.0">https://github.com/pinojs/sonic-boom/compare/v3.8.1...v4.0.0</a></p>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/pinojs/sonic-boom/commit/d860a5962106e4e1df60e307ddfe722305ff6b8b"><code>d860a59</code></a> Bumped v4.0.1</li>
<li><a href="https://github.com/pinojs/sonic-boom/commit/2068400bc01c9de330658cbcc20c382f4023d460"><code>2068400</code></a> ignored .tap in .npmignore;</li>
<li><a href="https://github.com/pinojs/sonic-boom/commit/f08dfddb26288412f848383c49ee0e8b744609d2"><code>f08dfdd</code></a> Bumped v4.0.0</li>
<li><a href="https://github.com/pinojs/sonic-boom/commit/696fe3c816617478336a13088ca1ed4df15e2e12"><code>696fe3c</code></a> Drop old node.js versions (<a href="https://redirect.github.com/pinojs/sonic-boom/issues/196">#196</a>)</li>
<li>See full diff in <a href="https://github.com/pinojs/sonic-boom/compare/v3.8.1...v4.0.1">compare view</a></li>
</ul>
</details>
<br />

<details>
<summary>Most Recent Ignore Conditions Applied to This Pull Request</summary>

| Dependency Name | Ignore Conditions |
| --- | --- |
| sonic-boom | [>= 2.a, < 3] |
</details>


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=sonic-boom&package-manager=npm_and_yarn&previous-version=3.8.1&new-version=4.0.1)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/browser.js b/browser.js
index 04b00ce..f786739 100644
--- a/browser.js
+++ b/browser.js
@@ -302,7 +302,7 @@ function createWrap (self, opts, rootLogger, level) {
       const proto = (Object.getPrototypeOf && Object.getPrototypeOf(this) === _console) ? _console : this
       for (var i = 0; i < args.length; i++) args[i] = arguments[i]
 
-      if (opts.serialize && !opts.asObject) {
+      if (opts.serialize && !opts.transmit) {
         applySerializers(args, this._serialize, this.serializers, this._stdErrSerialize)
       }
       if (opts.asObject || opts.formatters) {
@@ -330,17 +330,22 @@ function createWrap (self, opts, rootLogger, level) {
 
 function asObject (logger, level, args, ts, formatters = {}) {
   const {
-    level: levelFormatter = () => logger.levels.values[level],
+    level: levelFormatter,
     log: logObjectFormatter = (obj) => obj
   } = formatters
-  if (logger._serialize) applySerializers(args, logger._serialize, logger.serializers, logger._stdErrSerialize)
   const argsCloned = args.slice()
   let msg = argsCloned[0]
   const logObject = {}
   if (ts) {
     logObject.time = ts
   }
-  logObject.level = levelFormatter(level, logger.levels.values[level])
+
+  if (levelFormatter) {
+    const formattedLevel = levelFormatter(level, logger.levels.values[level])
+    Object.assign(logObject, formattedLevel)
+  } else {
+    logObject.level = logger.levels.values[level]
+  }
 
   let lvl = (logger._childLevel | 0) + 1
   if (lvl < 1) lvl = 1
@@ -361,9 +366,9 @@ function applySerializers (args, serialize, serializers, stdErrSerialize) {
   for (const i in args) {
     if (stdErrSerialize && args[i] instanceof Error) {
       args[i] = pino.stdSerializers.err(args[i])
-    } else if (typeof args[i] === 'object' && !Array.isArray(args[i])) {
+    } else if (typeof args[i] === 'object' && !Array.isArray(args[i]) && serialize) {
       for (const k in args[i]) {
-        if (serialize && serialize.indexOf(k) > -1 && k in serializers) {
+        if (serialize.indexOf(k) > -1 && k in serializers) {
           args[i][k] = serializers[k](args[i][k])
         }
       }
@@ -385,6 +390,7 @@ function transmit (logger, opts, args) {
     logger.serializers,
     logger._stdErrSerialize === undefined ? true : logger._stdErrSerialize
   )
+
   logger._logEvent.ts = ts
   logger._logEvent.messages = args.filter(function (arg) {
     // bindings can only be objects, so reference equality check via indexOf is fine
diff --git a/docs/api.md b/docs/api.md
index 1fc4384..80fa249 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -241,13 +241,13 @@ child.info('this will have both `foo: 1` and `bar: 2`')
 logger.info('this will still only have `foo: 1`')
 ```
 
-As of pino 7.x, when the `mixin` is used with the [`nestedKey` option](#opt-nestedkey), 
-the object returned from the `mixin` method will also be nested. Prior versions would mix 
-this object into the root. 
+As of pino 7.x, when the `mixin` is used with the [`nestedKey` option](#opt-nestedkey),
+the object returned from the `mixin` method will also be nested. Prior versions would mix
+this object into the root.
 
 ```js
 const logger = pino({
-    nestedKey: 'payload', 
+    nestedKey: 'payload',
     mixin() {
         return { requestId: requestId.currentId() }
     }
@@ -590,7 +590,7 @@ when using the `transport` option. In this case, an `Error` will be thrown.
 
 #### `onChild` (Function)
 
-The `onChild` function is a synchronous callback that will be called on each creation of a new child, passing the child instance as its first argument.  
+The `onChild` function is a synchronous callback that will be called on each creation of a new child, passing the child instance as its first argument.
 Any error thrown inside the callback will be uncaught and should be handled inside the callback.
 ```js
 const parent = require('pino')({ onChild: (instance) => {
@@ -609,7 +609,7 @@ Default: `pino.destination(1)` (STDOUT)
 The `destination` parameter can be a file descriptor, a file path, or an
 object with `dest` property pointing to a fd or path.
 An ordinary Node.js `stream` file descriptor can be passed as the
-destination (such as the result 
+destination (such as the result
 of `fs.createWriteStream`) but for peak log writing performance, it is strongly
 recommended to use `pino.destination` to create the destination stream.
 Note that the `destination` parameter can be the result of `pino.transport()`.
@@ -1001,7 +1001,7 @@ Adds to the bindings of this logger instance.
 **Note:** Does not overwrite bindings. Can potentially result in duplicate keys in
 log lines.
 
-* See [`bindings` parameter in `logger.child`](#logger-child-bindings)     
+* See [`bindings` parameter in `logger.child`](#logger-child-bindings)
 
 <a id="flush"></a>
 ### `logger.flush([cb])`
@@ -1239,6 +1239,30 @@ const transport = pino.transport({
 pino(transport)
 ```
 
+Multiple transports can now be defined to include pipelines:
+
+```js
+const pino = require('pino')
+const transport = pino.transport({
+  targets: [{
+    level: 'info',
+    target: 'pino-pretty' // must be installed separately
+  }, {
+    level: 'trace',
+    target: 'pino/file',
+    options: { destination: '/path/to/store/logs' }
+  }, {
+    pipeline: [{
+      target: 'pino-syslog' // must be installed separately
+    }, {
+      target: 'pino-socket' // must be installed separately
+    }]
+  }
+  ]
+})
+pino(transport)
+```
+
 If `WeakRef`, `WeakMap`, and `FinalizationRegistry` are available in the current runtime (v14.5.0+), then the thread
 will be automatically terminated in case the stream or logger goes out of scope.
 The `transport()` function adds a listener to `process.on('beforeExit')` and `process.on('exit')` to ensure the worker
@@ -1276,7 +1300,7 @@ For more on transports, how they work, and how to create them see the [`Transpor
 * `target`:  The transport to pass logs through. This may be an installed module name or an absolute path.
 * `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.
+* `targets`: May be specified instead of `target`. Must be an array of transport configurations and/or pipelines. 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.
 * `pipeline`: May be specified instead of `target`. Must be an array of transport configurations. Transport configurations include the aforementioned `options` and `target` options. All intermediate steps in the pipeline _must_ be `Transform` streams and not `Writable`.
 * `dedupe`: See [pino.multistream options](#pino-multistream)
 
diff --git a/docs/browser.md b/docs/browser.md
index deb465c..a23ce83 100644
--- a/docs/browser.md
+++ b/docs/browser.md
@@ -144,7 +144,7 @@ Unlike server pino the serializers apply to every object passed to the logger me
 if the `asObject` option is `true`, this results in the serializers applying to the
 first object (as in server pino).
 
-For more info on serializers see https://github.com/pinojs/pino/blob/master/docs/api.md#parameters.
+For more info on serializers see https://github.com/pinojs/pino/blob/master/docs/api.md#mergingobject.
 
 ### `transmit` (Object)
 
diff --git a/docs/bundling.md b/docs/bundling.md
index 6467b8e..1c326a9 100644
--- a/docs/bundling.md
+++ b/docs/bundling.md
@@ -7,7 +7,6 @@ In particular, a bundler must ensure that the following files are also bundled s
 * `lib/worker.js` from the `thread-stream` dependency
 * `file.js`
 * `lib/worker.js`
-* `lib/worker-pipeline.js`
 * Any transport used by the user (like `pino-pretty`)
 
 Once the files above have been generated, the bundler must also add information about the files above by injecting a code that sets `__bundlerPathsOverrides` in the `globalThis` object.
@@ -22,12 +21,11 @@ globalThis.__bundlerPathsOverrides = {
   'thread-stream-worker': pinoWebpackAbsolutePath('./thread-stream-worker.js')
   'pino/file': pinoWebpackAbsolutePath('./pino-file.js'),
   'pino-worker': pinoWebpackAbsolutePath('./pino-worker.js'),
-  'pino-pipeline-worker': pinoWebpackAbsolutePath('./pino-pipeline-worker.js'),
   'pino-pretty': pinoWebpackAbsolutePath('./pino-pretty.js'),
 };
 ```
 
-Note that `pino/file`, `pino-worker`, `pino-pipeline-worker`, and `thread-stream-worker` are required identifiers. Other identifiers are possible based on the user configuration.
+Note that `pino/file`, `pino-worker` and `thread-stream-worker` are required identifiers. Other identifiers are possible based on the user configuration.
 
 ## Webpack Plugin
 
diff --git a/docs/web.md b/docs/web.md
index 754e033..49a31d0 100644
--- a/docs/web.md
+++ b/docs/web.md
@@ -236,24 +236,34 @@ See the [nestjs-pino README](https://npm.im/nestjs-pino) for more info.
 ## Pino with H3
 
 ```sh
-npm install pino-http
+npm install pino-http h3
 ```
 
+Save as `server.mjs`:
+
 ```js
-import { createServer } from 'http'
-import { createApp } from 'h3'
+import { createApp, createRouter, eventHandler, fromNodeMiddleware } from "h3";
 import pino from 'pino-http'
 
-const app = createApp()
+export const app = createApp();
 
-app.use(pino())
+const router = createRouter();
+app.use(router);
+app.use(fromNodeMiddleware(pino()))
 
-app.use('/', (req) => {
-  req.log.info('something')
+app.use(eventHandler((event) => {
+  event.node.req.log.info('something')
   return 'hello world'
-})
-
-createServer(app).listen(process.env.PORT || 3000)
+}))
+
+router.get(
+  "/",
+  eventHandler((event) => {
+    return { path: event.path, message: "Hello World!" };
+  }),
+);
 ```
 
+Execute `npx --yes listhen -w --open ./server.mjs`.
+
 See the [pino-http README](https://npm.im/pino-http) for more info.
diff --git a/lib/transport-stream.js b/lib/transport-stream.js
index b152fc9..da864a9 100644
--- a/lib/transport-stream.js
+++ b/lib/transport-stream.js
@@ -13,7 +13,7 @@ module.exports = loadTransportStreamBuilder
 async function loadTransportStreamBuilder (target) {
   let fn
   try {
-    const toLoad = 'file://' + target
+    const toLoad = target.startsWith('file://') ? target : 'file://' + target
 
     if (toLoad.endsWith('.ts') || toLoad.endsWith('.cts')) {
       // TODO: add support for the TSM modules loader ( https://github.com/lukeed/tsm ).
diff --git a/lib/transport.js b/lib/transport.js
index 6be075f..b8ce347 100644
--- a/lib/transport.js
+++ b/lib/transport.js
@@ -87,20 +87,29 @@ function transport (fullOptions) {
 
   if (targets) {
     target = bundlerOverrides['pino-worker'] || join(__dirname, 'worker.js')
-    options.targets = targets.map((dest) => {
+    options.targets = targets.filter(dest => dest.target).map((dest) => {
       return {
         ...dest,
         target: fixTarget(dest.target)
       }
     })
+    options.pipelines = targets.filter(dest => dest.pipeline).map((dest) => {
+      return dest.pipeline.map((t) => {
+        return {
+          ...t,
+          level: dest.level, // duplicate the pipeline `level` property defined in the upper level
+          target: fixTarget(t.target)
+        }
+      })
+    })
   } else if (pipeline) {
-    target = bundlerOverrides['pino-pipeline-worker'] || join(__dirname, 'worker-pipeline.js')
-    options.targets = pipeline.map((dest) => {
+    target = bundlerOverrides['pino-worker'] || join(__dirname, 'worker.js')
+    options.pipelines = [pipeline.map((dest) => {
       return {
         ...dest,
         target: fixTarget(dest.target)
       }
-    })
+    })]
   }
 
   if (levels) {
diff --git a/lib/worker-pipeline.js b/lib/worker-pipeline.js
deleted file mode 100644
index 76cb3b8..0000000
--- a/lib/worker-pipeline.js
+++ /dev/null
@@ -1,38 +0,0 @@
-'use strict'
-
-const EE = require('events')
-const loadTransportStreamBuilder = require('./transport-stream')
-const { pipeline, PassThrough } = require('stream')
-
-// This file is not checked by the code coverage tool,
-// as it is not reliable.
-
-/* istanbul ignore file */
-
-module.exports = async function ({ targets }) {
-  const streams = await Promise.all(targets.map(async (t) => {
-    const fn = await loadTransportStreamBuilder(t.target)
-    const stream = await fn(t.options)
-    return stream
-  }))
-  const ee = new EE()
-
-  const stream = new PassThrough({
-    autoDestroy: true,
-    destroy (_, cb) {
-      ee.on('error', cb)
-      ee.on('closed', cb)
-    }
-  })
-
-  pipeline(stream, ...streams, function (err) {
-    if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
-      ee.emit('error', err)
-      return
-    }
-
-    ee.emit('closed')
-  })
-
-  return stream
-}
diff --git a/lib/worker.js b/lib/worker.js
index c20c19a..dcbd5d3 100644
--- a/lib/worker.js
+++ b/lib/worker.js
@@ -1,5 +1,7 @@
 'use strict'
 
+const EE = require('events')
+const { pipeline, PassThrough } = require('stream')
 const pino = require('../pino.js')
 const build = require('pino-abstract-transport')
 const loadTransportStreamBuilder = require('./transport-stream')
@@ -9,36 +11,144 @@ const loadTransportStreamBuilder = require('./transport-stream')
 
 /* istanbul ignore file */
 
-module.exports = async function ({ targets, levels, dedupe }) {
-  targets = await Promise.all(targets.map(async (t) => {
-    const fn = await loadTransportStreamBuilder(t.target)
-    const stream = await fn(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()
+/*
+ * > Multiple targets & pipelines
+ *
+ *
+ * ┌─────────────────────────────────────────────────┐    ┌─────┐
+ * │                                                 │    │  p  │
+ * │                                                 │    │  i  │
+ * │                   target                        │    │  n  │
+ * │               │ ────────────────────────────────┼────┤  o  │
+ * │   targets     │   target                        │    │  .  │
+ * │ ────────────► │ ────────────────────────────────┼────┤  m  │       source
+ * │               │   target                        │    │  u  │         │
+ * │               │ ────────────────────────────────┼────┤  l  │         │write
+ * │               │                                 │    │  t  │         ▼
+ * │               │  pipeline   ┌───────────────┐   │    │  i  │      ┌────────┐
+ * │               │ ──────────► │  PassThrough  ├───┼────┤  s  ├──────┤        │
+ * │               │             └───────────────┘   │    │  t  │ write│ Thread │
+ * │               │                                 │    │  r  │◄─────┤ Stream │
+ * │               │  pipeline   ┌───────────────┐   │    │  e  │      │        │
+ * │               │ ──────────► │  PassThrough  ├───┼────┤  a  │      └────────┘
+ * │                             └───────────────┘   │    │  m  │
+ * │                                                 │    │     │
+ * └─────────────────────────────────────────────────┘    └─────┘
+ *
+ *
+ *
+ *  > One single pipeline or target
+ *
+ *
+ *                                                           source
+ *                                                             │
+ * ┌────────────────────────────────────────────────┐          │write
+ * │                                                │          ▼
+ * │                                                │      ┌────────┐
+ * │   targets     │   target                       │      │        │
+ * │ ────────────► │  ──────────────────────────────┤      │        │
+ * │               │                                │      │        │
+ * │                                                ├──────┤        │
+ * │                                                │      │        │
+ * │                                                │      │        │
+ * │                     OR                         │      │        │
+ * │                                                │      │        │
+ * │                                                │      │        │
+ * │                               ┌──────────────┐ │      │        │
+ * │   targets     │   pipeline    │              │ │      │ Thread │
+ * │ ────────────► │  ────────────►│ PassThrough  ├─┤      │ Stream │
+ * │               │               │              │ │      │        │
+ * │                               └──────────────┘ │      │        │
+ * │                                                │      │        │
+ * │                     OR                         │ write│        │
+ * │                                                │◄─────┤        │
+ * │                                                │      │        │
+ * │                ┌──────────────┐                │      │        │
+ * │    pipeline    │              │                │      │        │
+ * │ ──────────────►│ PassThrough  ├────────────────┤      │        │
+ * │                │              │                │      │        │
+ * │                └──────────────┘                │      └────────┘
+ * │                                                │
+ * │                                                │
+ * └────────────────────────────────────────────────┘
+ */
+
+module.exports = async function ({ targets, pipelines, levels, dedupe }) {
+  const targetStreams = []
+
+  // Process targets
+  if (targets && targets.length) {
+    targets = await Promise.all(targets.map(async (t) => {
+      const fn = await loadTransportStreamBuilder(t.target)
+      const stream = await fn(t.options)
+      return {
+        level: t.level,
+        stream
       }
+    }))
+
+    targetStreams.push(...targets)
+  }
+
+  // Process pipelines
+  if (pipelines && pipelines.length) {
+    pipelines = await Promise.all(
+      pipelines.map(async (p) => {
+        let level
+        const pipeDests = await Promise.all(
+          p.map(async (t) => {
+            // level assigned to pipeline is duplicated over all its targets, just store it
+            level = t.level
+            const fn = await loadTransportStreamBuilder(t.target)
+            const stream = await fn(t.options)
+            return stream
+          }
+          ))
 
-      function closeCb () {
-        if (--expected === 0) {
-          cb(err)
+        return {
+          level,
+          stream: createPipeline(pipeDests)
+        }
+      })
+    )
+    targetStreams.push(...pipelines)
+  }
+
+  // Skip building the multistream step if either one single pipeline or target is defined and
+  // return directly the stream instance back to TreadStream.
+  // This is equivalent to define either:
+  //
+  // pino.transport({ target: ... })
+  //
+  // OR
+  //
+  // pino.transport({ pipeline: ... })
+  if (targetStreams.length === 1) {
+    return targetStreams[0].stream
+  } else {
+    return build(process, {
+      parse: 'lines',
+      metadata: true,
+      close (err, cb) {
+        let expected = 0
+        for (const transport of targetStreams) {
+          expected++
+          transport.stream.on('close', closeCb)
+          transport.stream.end()
+        }
+
+        function closeCb () {
+          if (--expected === 0) {
+            cb(err)
+          }
         }
       }
-    }
-  })
+    })
+  }
 
+  // TODO: Why split2 was not used for pipelines?
   function process (stream) {
-    const multi = pino.multistream(targets, { levels, dedupe })
+    const multi = pino.multistream(targetStreams, { levels, dedupe })
     // TODO manage backpressure
     stream.on('data', function (chunk) {
       const { lastTime, lastMsg, lastObj, lastLevel } = this
@@ -51,4 +161,34 @@ module.exports = async function ({ targets, levels, dedupe }) {
       multi.write(chunk + '\n')
     })
   }
+
+  /**
+ * Creates a pipeline using the provided streams and return an instance of `PassThrough` stream
+ * as a source for the pipeline.
+ *
+ * @param {(TransformStream|WritableStream)[]} streams An array of streams.
+ *   All intermediate streams in the array *MUST* be `Transform` streams and only the last one `Writable`.
+ * @returns A `PassThrough` stream instance representing the source stream of the pipeline
+ */
+  function createPipeline (streams) {
+    const ee = new EE()
+    const stream = new PassThrough({
+      autoDestroy: true,
+      destroy (_, cb) {
+        ee.on('error', cb)
+        ee.on('closed', cb)
+      }
+    })
+
+    pipeline(stream, ...streams, function (err) {
+      if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
+        ee.emit('error', err)
+        return
+      }
+
+      ee.emit('closed')
+    })
+
+    return stream
+  }
 }
diff --git a/package.json b/package.json
index 76bf66b..68fef31 100644
--- a/package.json
+++ b/package.json
@@ -84,7 +84,7 @@
     "log": "^6.0.0",
     "loglevel": "^1.6.7",
     "midnight-smoker": "1.1.1",
-    "pino-pretty": "^10.2.1",
+    "pino-pretty": "^11.0.0",
     "pre-commit": "^1.2.2",
     "proxyquire": "^2.1.3",
     "pump": "^3.0.0",
@@ -111,7 +111,7 @@
     "quick-format-unescaped": "^4.0.3",
     "real-require": "^0.2.0",
     "safe-stable-stringify": "^2.3.1",
-    "sonic-boom": "^3.7.0",
+    "sonic-boom": "^4.0.1",
     "thread-stream": "^2.6.0"
   },
   "tsd": {
diff --git a/pino.d.ts b/pino.d.ts
index b6709a5..3efe722 100644
--- a/pino.d.ts
+++ b/pino.d.ts
@@ -259,10 +259,11 @@ declare namespace pino {
 
     interface TransportPipelineOptions<TransportOptions = Record<string, any>> extends TransportBaseOptions<TransportOptions>{
         pipeline: TransportSingleOptions<TransportOptions>[]
+        level?: LevelWithSilentOrString
     }
 
     interface TransportMultiOptions<TransportOptions = Record<string, any>> extends TransportBaseOptions<TransportOptions>{
-        targets: readonly TransportTargetOptions<TransportOptions>[],
+        targets: readonly (TransportTargetOptions<TransportOptions>|TransportPipelineOptions<TransportOptions>)[],
         levels?: Record<string, number>
         dedupe?: boolean
     }


Test patch:
diff --git a/test/browser-serializers.test.js b/test/browser-serializers.test.js
index 77d6dee..07cfa60 100644
--- a/test/browser-serializers.test.js
+++ b/test/browser-serializers.test.js
@@ -284,7 +284,6 @@ test('children inherit parent serializers', ({ end, is }) => {
 
 test('children serializers get called', ({ end, is }) => {
   const parent = pino({
-    test: 'this',
     browser: {
       serialize: true,
       write (o) {
@@ -301,7 +300,6 @@ test('children serializers get called', ({ end, is }) => {
 
 test('children serializers get called when inherited from parent', ({ end, is }) => {
   const parent = pino({
-    test: 'this',
     serializers: parentSerializers,
     browser: {
       serialize: true,
diff --git a/test/browser.test.js b/test/browser.test.js
index c296411..ab0b045 100644
--- a/test/browser.test.js
+++ b/test/browser.test.js
@@ -166,7 +166,7 @@ test('opts.browser.asObject logs pino-like object to console', ({ end, ok, is })
   end()
 })
 
-test('opts.browser.formatters logs pino-like object to console', ({ end, ok, is }) => {
+test('opts.browser.formatters (level) logs pino-like object to console', ({ end, ok, is }) => {
   const info = console.info
   console.info = function (o) {
     is(o.level, 30)
@@ -189,7 +189,7 @@ test('opts.browser.formatters logs pino-like object to console', ({ end, ok, is
   end()
 })
 
-test('opts.browser.formatters logs pino-like object to console', ({ end, ok, is }) => {
+test('opts.browser.formatters (log) logs pino-like object to console', ({ end, ok, is }) => {
   const info = console.info
   console.info = function (o) {
     is(o.level, 30)
@@ -213,40 +213,66 @@ test('opts.browser.formatters logs pino-like object to console', ({ end, ok, is
   end()
 })
 
-test('opts.browser.write func log single string', ({ end, ok, is }) => {
-  const instance = pino({
+test('opts.browser.serialize and opts.browser.transmit only serializes log data once', ({ end, ok, is }) => {
+  const instance = require('../browser')({
+    serializers: {
+      extras (data) {
+        return { serializedExtras: data }
+      }
+    },
     browser: {
+      serialize: ['extras'],
+      transmit: {
+        level: 'info',
+        send (level, o) {
+          is(o.messages[0].extras.serializedExtras, 'world')
+        }
+      }
+    }
+  })
+
+  instance.info({ extras: 'world' }, 'test')
+  end()
+})
+
+test('opts.browser.serialize and opts.asObject only serializes log data once', ({ end, ok, is }) => {
+  const instance = require('../browser')({
+    serializers: {
+      extras (data) {
+        return { serializedExtras: data }
+      }
+    },
+    browser: {
+      serialize: ['extras'],
+      asObject: true,
       write: function (o) {
-        is(o.level, 30)
-        is(o.msg, 'test')
-        ok(o.time)
+        is(o.extras.serializedExtras, 'world')
       }
     }
   })
-  instance.info('test')
 
+  instance.info({ extras: 'world' }, 'test')
   end()
 })
 
-test('opts.browser.write func string joining', ({ end, ok, is }) => {
+test('opts.browser.write func log single string', ({ end, ok, is }) => {
   const instance = pino({
     browser: {
       write: function (o) {
         is(o.level, 30)
-        is(o.msg, 'test test2 test3')
+        is(o.msg, 'test')
         ok(o.time)
       }
     }
   })
-  instance.info('test %s %s', 'test2', 'test3')
+  instance.info('test')
 
   end()
 })
 
-test('opts.browser.write func string joining when asObject is true', ({ end, ok, is }) => {
+test('opts.browser.write func string joining', ({ end, ok, is }) => {
   const instance = pino({
     browser: {
-      asObject: true,
       write: function (o) {
         is(o.level, 30)
         is(o.msg, 'test test2 test3')
diff --git a/test/transport/bundlers-support.test.js b/test/transport/bundlers-support.test.js
index efe8a5e..cb10221 100644
--- a/test/transport/bundlers-support.test.js
+++ b/test/transport/bundlers-support.test.js
@@ -95,33 +95,3 @@ test('pino.transport with worker destination overridden by bundler and mjs trans
 
   globalThis.__bundlerPathsOverrides = undefined
 })
-
-test('pino.transport with worker-pipeline destination overridden by bundler', async ({ same, teardown }) => {
-  globalThis.__bundlerPathsOverrides = {
-    'pino-pipeline-worker': join(__dirname, '..', '..', 'lib/worker-pipeline.js')
-  }
-
-  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'
-  })
-
-  globalThis.__bundlerPathsOverrides = undefined
-})
diff --git a/test/transport/core.test.js b/test/transport/core.test.js
index ee5ef10..b8af673 100644
--- a/test/transport/core.test.js
+++ b/test/transport/core.test.js
@@ -134,7 +134,7 @@ test('pino.transport with two files', async ({ same, teardown }) => {
   const transport = pino.transport({
     targets: [{
       level: 'info',
-      target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+      target: 'file://' + join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
       options: { destination: dest1 }
     }, {
       level: 'info',
diff --git a/test/transport/pipeline.test.js b/test/transport/pipeline.test.js
index 25b244e..a845d44 100644
--- a/test/transport/pipeline.test.js
+++ b/test/transport/pipeline.test.js
@@ -6,6 +6,7 @@ const { readFile } = require('fs').promises
 const { watchFileCreated, file } = require('../helper')
 const { test } = require('tap')
 const pino = require('../../')
+const { DEFAULT_LEVELS } = require('../../lib/constants')
 
 const { pid } = process
 const hostname = os.hostname()
@@ -29,8 +30,106 @@ test('pino.transport with a pipeline', async ({ same, teardown }) => {
   same(result, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello',
     service: 'pino' // this property was added by the transform
   })
 })
+
+test('pino.transport with targets containing pipelines', async ({ same, teardown }) => {
+  const destinationA = file()
+  const destinationB = file()
+  const transport = pino.transport({
+    targets: [
+      {
+        target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+        options: { destination: destinationA }
+      },
+      {
+        pipeline: [
+          {
+            target: join(__dirname, '..', 'fixtures', 'transport-transform.js')
+          },
+          {
+            target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+            options: { destination: destinationB }
+          }
+        ]
+      }
+    ]
+  })
+
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello')
+  await watchFileCreated(destinationA)
+  await watchFileCreated(destinationB)
+  const resultA = JSON.parse(await readFile(destinationA))
+  const resultB = JSON.parse(await readFile(destinationB))
+  delete resultA.time
+  delete resultB.time
+  same(resultA, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.info,
+    msg: 'hello'
+  })
+  same(resultB, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.info,
+    msg: 'hello',
+    service: 'pino' // this property was added by the transform
+  })
+})
+
+test('pino.transport with targets containing pipelines with levels defined and dedupe', async ({ same, teardown }) => {
+  const destinationA = file()
+  const destinationB = file()
+  const transport = pino.transport({
+    targets: [
+      {
+        target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+        options: { destination: destinationA },
+        level: DEFAULT_LEVELS.info
+      },
+      {
+        pipeline: [
+          {
+            target: join(__dirname, '..', 'fixtures', 'transport-transform.js')
+          },
+          {
+            target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+            options: { destination: destinationB }
+          }
+        ],
+        level: DEFAULT_LEVELS.error
+      }
+    ],
+    dedupe: true
+  })
+
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello info')
+  instance.error('hello error')
+  await watchFileCreated(destinationA)
+  await watchFileCreated(destinationB)
+  const resultA = JSON.parse(await readFile(destinationA))
+  const resultB = JSON.parse(await readFile(destinationB))
+  delete resultA.time
+  delete resultB.time
+  same(resultA, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.info,
+    msg: 'hello info'
+  })
+  same(resultB, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.error,
+    msg: 'hello error',
+    service: 'pino' // this property was added by the transform
+  })
+})