Wait for destination to close (#533)

* Drop pump & readable-stream Signed-off-by: Matteo Collina <[email protected]> * wait for destination to close Signed-off-by: Matteo Collina <[email protected]> * drop old nodes Signed-off-by: Matteo Collina <[email protected]> * Revert "drop old nodes" This reverts commit 01d413d. * Revert "Drop pump & readable-stream" This reverts commit a2738a5. Signed-off-by: Matteo Collina <[email protected]> --------- Signed-off-by: Matteo Collina <[email protected]>
pinojs · Oct 15, 2024 · 70ade71 · 70ade71
1 parent ba1e844
commit 70ade71
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 3 deletions.
diff --git a/index.js b/index.js
@@ -132,6 +132,7 @@ function prettyFactory (options) {
  */
 function build (opts = {}) {
   let pretty = prettyFactory(opts)
+  let destination
   return abstractTransport(function (source) {
     source.on('message', function pinoConfigListener (message) {
       if (!message || message.code !== 'PINO_CONFIG') return
@@ -152,8 +153,6 @@ function build (opts = {}) {
       }
     })
 
-    let destination
-
     if (typeof opts.destination === 'object' && typeof opts.destination.write === 'function') {
       destination = opts.destination
     } else {
@@ -171,7 +170,14 @@ function build (opts = {}) {
 
     pump(source, stream, destination)
     return stream
-  }, { parse: 'lines' })
+  }, {
+    parse: 'lines',
+    close (err, cb) {
+      destination.on('close', () => {
+        cb(err)
+      })
+    }
+  })
 }
 
 module.exports = build

diff --git a/test/basic.test.js b/test/basic.test.js
@@ -1098,6 +1098,38 @@ test('basic prettifier tests', (t) => {
     t.equal(closeCalled, false)
   })
 
+  t.test('wait for close event from destination', (t) => {
+    t.plan(2)
+    const destination = pino.destination({ minLength: 4096, sync: true })
+    const prettyDestination = pinoPretty({ destination, colorize: false })
+    const log = pino(prettyDestination)
+    log.info('this message has been buffered')
+    const chunks = []
+    const { close, writeSync } = fs
+    fs.close = new Proxy(close, {
+      apply: (target, self, args) => {
+      }
+    })
+    fs.writeSync = new Proxy(writeSync, {
+      apply: (target, self, args) => {
+        chunks.push(args[1])
+        return args[1].length
+      }
+    })
+    t.teardown(() => {
+      Object.assign(fs, { close, writeSync })
+    })
+    let destinationClosed = false
+    destination.on('close', () => {
+      destinationClosed = true
+    })
+    prettyDestination.on('close', () => {
+      t.match(chunks.join(''), /INFO .+: this message has been buffered/)
+      t.equal(destinationClosed, true)
+    })
+    prettyDestination.end()
+  })
+
   t.test('stream usage', async (t) => {
     t.plan(1)
     const tmpDir = path.join(__dirname, '.tmp_' + Date.now())