refactor: update implementation

lib/reply.js

@@ -49,7 +49,6 @@ const {
   FST_ERR_MISSING_CONTENTTYPE_SERIALIZATION_FN
 } = require('./errors')
 const warning = require('./warnings')
-const { nextTick } = require('process')
 
 function Reply (res, request, log) {
   this.raw = res
@@ -620,10 +619,6 @@ function sendStream (payload, res, reply) {
   let sourceOpen = true
   let errorLogged = false
 
-  // we should clone the payload for trailer
-  const stream = Cloneable(payload)
-  sendStreamTrailer(stream, res, reply)
-
   eos(payload, { readable: true, writable: false }, function (err) {
     sourceOpen = false
     if (err != null) {
@@ -670,15 +665,13 @@ function sendStream (payload, res, reply) {
     reply.log.warn('response will send, but you shouldn\'t use res.writeHead in stream mode')
   }
   if (reply[kReplyTrailers] === null) {
-    stream.pipe(res)
+    payload.pipe(res)
   } else {
-    // we should not pipe to res when using trailer
-    // it will immediately close the response after
-    // the payload is drain
-    stream.on('data', function (chunk, encoding) {
-      console.log(chunk, encoding)
-      res.write(chunk, encoding)
-    })
+    // we should clone the payload for trailer
+    const stream = Cloneable(payload)
+    sendStreamTrailer(stream, res, reply)
+
+    stream.pipe(res, { end: false })
   }
 }
 
@@ -699,13 +692,13 @@ function sendTrailer (payload, res, reply) {
       if (handled === 0) {
         // send trailer when all handlers are resolved
         res.addTrailers(trailers)
-        // we need to properly end the response
-        nextTick(() => res.end())
+        // we need to end the response properly
+        res.end()
       }
     }
     const result = reply[kReplyTrailers][trailerName](reply, payload instanceof Cloneable ? payload.clone() : payload, cb)
     if (typeof result === 'object' && typeof result.then === 'function') {
-      result.then(cb, cb)
+      result.then((v) => cb(null, v), cb)
     } else if (result !== undefined && result !== null) {
       // fallback
       // TODO: should be deprecated

test/reply-trailers.test.js

@@ -5,6 +5,8 @@ const test = t.test
 const Fastify = require('..')
 const { Readable } = require('stream')
 const { createHash } = require('crypto')
+const { promisify } = require('util')
+const sleep = promisify(setTimeout)
 
 test('send trailers when payload is empty string', t => {
   t.plan(5)
@@ -141,7 +143,7 @@ test('send trailers when payload is stream', ({ test, plan }) => {
     })
   })
 
-  test('pipe to cloned stream', t => {
+  test('single trailers', t => {
     t.plan(8)
 
     const fastify = Fastify()
@@ -175,7 +177,7 @@ test('send trailers when payload is stream', ({ test, plan }) => {
     })
   })
 
-  test('pipe multiple times', t => {
+  test('multiple trailers', t => {
     t.plan(11)
 
     const fastify = Fastify()
@@ -222,6 +224,38 @@ test('send trailers when payload is stream', ({ test, plan }) => {
   })
 })
 
+test('async-await trailers', (t) => {
+  t.plan(7)
+
+  const fastify = Fastify()
+  t.teardown(fastify.close.bind(fastify))
+  const data = JSON.stringify({ hello: 'world' })
+  const hash = createHash('md5')
+  hash.update(data)
+  const md5 = hash.digest('hex')
+
+  fastify.get('/', function (request, reply) {
+    reply.trailer('Content-MD5', async function (reply, payload) {
+      t.equal(data, payload)
+      await sleep(500)
+      return md5
+    })
+    reply.send(data)
+  })
+
+  fastify.inject({
+    method: 'GET',
+    url: '/'
+  }, (error, res) => {
+    t.error(error)
+    t.equal(res.statusCode, 200)
+    t.equal(res.headers['transfer-encoding'], 'chunked')
+    t.equal(res.headers.trailer, 'content-md5')
+    t.equal(res.trailers['content-md5'], md5)
+    t.notHas(res.headers, 'content-length')
+  })
+})
+
 test('removeTrailer', t => {
   t.plan(6)