Update checked-in dependencies

node_modules/nock/lib/playback_interceptor.js

@@ -1,10 +1,10 @@
 'use strict'
 
+const stream = require('stream')
 const util = require('util')
 const zlib = require('zlib')
 const debug = require('debug')('nock.playback_interceptor')
 const common = require('./common')
-const DelayedBody = require('./delayed_body')
 
 function parseJSONRequestBody(req, requestBody) {
   if (!requestBody || !common.isJSONContent(req.headers)) {
@@ -71,6 +71,44 @@ function selectDefaultHeaders(existingHeaders, defaultHeaders) {
   return result
 }
 
+// Presents a list of Buffers as a Readable
+class ReadableBuffers extends stream.Readable {
+  constructor(buffers, opts = {}) {
+    super(opts)
+
+    this.buffers = buffers
+  }
+
+  _read(size) {
+    while (this.buffers.length) {
+      if (!this.push(this.buffers.shift())) {
+        return
+      }
+    }
+    this.push(null)
+  }
+}
+
+function convertBodyToStream(body) {
+  if (common.isStream(body)) {
+    return body
+  }
+
+  if (body === undefined) {
+    return new ReadableBuffers([])
+  }
+
+  if (Buffer.isBuffer(body)) {
+    return new ReadableBuffers([body])
+  }
+
+  if (typeof body !== 'string') {
+    body = JSON.stringify(body)
+  }
+
+  return new ReadableBuffers([Buffer.from(body)])
+}
+
 /**
  * Play back an interceptor using the given request and mock response.
  */
@@ -83,28 +121,23 @@ function playbackInterceptor({
   response,
   interceptor,
 }) {
-  function emitError(error) {
-    process.nextTick(() => {
-      req.emit('error', error)
-    })
-  }
+  const { logger } = interceptor.scope
 
   function start() {
-    interceptor.req = req
     req.headers = req.getHeaders()
 
     interceptor.scope.emit('request', req, interceptor, requestBodyString)
 
     if (typeof interceptor.errorMessage !== 'undefined') {
-      interceptor.markConsumed()
-
       let error
       if (typeof interceptor.errorMessage === 'object') {
         error = interceptor.errorMessage
       } else {
         error = new Error(interceptor.errorMessage)
       }
-      common.setTimeout(() => emitError(error), interceptor.getTotalDelay())
+
+      const delay = interceptor.delayBodyInMs + interceptor.delayConnectionInMs
+      common.setTimeout(() => req.destroy(error), delay)
       return
     }
 
@@ -114,7 +147,14 @@ function playbackInterceptor({
 
     // Clone headers/rawHeaders to not override them when evaluating later
     response.rawHeaders = [...interceptor.rawHeaders]
-    debug('response.rawHeaders:', response.rawHeaders)
+    logger('response.rawHeaders:', response.rawHeaders)
+
+    // TODO: MAJOR: Don't tack the request onto the interceptor.
+    // The only reason we do this is so that it's available inside reply functions.
+    // It would be better to pass the request as an argument to the functions instead.
+    // Not adding the req as a third arg now because it should first be decided if (path, body, req)
+    // is the signature we want to go with going forward.
+    interceptor.req = req
 
     if (interceptor.replyFunction) {
       const parsedRequestBody = parseJSONRequestBody(req, requestBodyString)
@@ -128,8 +168,8 @@ function playbackInterceptor({
       // At this point `fn` is either a synchronous function or a promise-returning function;
       // wrapping in `Promise.resolve` makes it into a promise either way.
       Promise.resolve(fn.call(interceptor, options.path, parsedRequestBody))
-        .then(responseBody => continueWithResponseBody({ responseBody }))
-        .catch(err => emitError(err))
+        .then(continueWithResponseBody)
+        .catch(err => req.destroy(err))
       return
     }
 
@@ -142,8 +182,8 @@ function playbackInterceptor({
       }
 
       Promise.resolve(fn.call(interceptor, options.path, parsedRequestBody))
-        .then(fullReplyResult => continueWithFullResponse({ fullReplyResult }))
-        .catch(err => emitError(err))
+        .then(continueWithFullResponse)
+        .catch(err => req.destroy(err))
       return
     }
 
@@ -155,21 +195,12 @@ function playbackInterceptor({
       //  of response buffers which should be mocked one by one.
       //  (otherwise decompressions after the first one fails as unzip expects to receive
       //  buffer by buffer and not one single merged buffer)
-
-      if (interceptor.delayInMs) {
-        emitError(
-          new Error(
-            'Response delay of the body is currently not supported with content-encoded responses.'
-          )
-        )
-        return
-      }
-
       const bufferData = Array.isArray(interceptor.body)
         ? interceptor.body
         : [interceptor.body]
       const responseBuffers = bufferData.map(data => Buffer.from(data, 'hex'))
-      continueWithResponseBody({ responseBuffers })
+      const responseBody = new ReadableBuffers(responseBuffers)
+      continueWithResponseBody(responseBody)
       return
     }
 
@@ -194,155 +225,103 @@ function playbackInterceptor({
       }
     }
 
-    return continueWithResponseBody({ responseBody })
+    return continueWithResponseBody(responseBody)
   }
 
-  function continueWithFullResponse({ fullReplyResult }) {
+  function continueWithFullResponse(fullReplyResult) {
     let responseBody
     try {
       responseBody = parseFullReplyResult(response, fullReplyResult)
-    } catch (innerErr) {
-      emitError(innerErr)
+    } catch (err) {
+      req.destroy(err)
       return
     }
 
-    continueWithResponseBody({ responseBody })
+    continueWithResponseBody(responseBody)
   }
 
-  function continueWithResponseBody({ responseBuffers, responseBody }) {
-    //  Transform the response body if it exists (it may not exist
-    //  if we have `responseBuffers` instead)
-    if (responseBody !== undefined) {
-      debug('transform the response body')
+  function prepareResponseHeaders(body) {
+    const defaultHeaders = [...interceptor.scope._defaultReplyHeaders]
 
-      if (interceptor.delayInMs) {
-        debug(
-          'delaying the response for',
-          interceptor.delayInMs,
-          'milliseconds'
-        )
-        // Because setTimeout is called immediately in DelayedBody(), so we
-        // need count in the delayConnectionInMs.
-        responseBody = new DelayedBody(
-          interceptor.getTotalDelay(),
-          responseBody
-        )
-      }
+    // Include a JSON content type when JSON.stringify is called on the body.
+    // This is a convenience added by Nock that has no analog in Node. It's added to the
+    // defaults, so it will be ignored if the caller explicitly provided the header already.
+    const isJSON =
+      body !== undefined &&
+      typeof body !== 'string' &&
+      !Buffer.isBuffer(body) &&
+      !common.isStream(body)
 
-      if (common.isStream(responseBody)) {
-        debug('response body is a stream')
-        responseBody.pause()
-        responseBody.on('data', function(d) {
-          response.push(d)
-        })
-        responseBody.on('end', function() {
-          response.push(null)
-          // https://nodejs.org/dist/latest-v10.x/docs/api/http.html#http_message_complete
-          response.complete = true
-        })
-        responseBody.on('error', function(err) {
-          response.emit('error', err)
-        })
-      } else if (!Buffer.isBuffer(responseBody)) {
-        if (typeof responseBody === 'string') {
-          responseBody = Buffer.from(responseBody)
-        } else {
-          responseBody = JSON.stringify(responseBody)
-          response.rawHeaders.push('Content-Type', 'application/json')
-        }
-      }
-      // Why are strings converted to a Buffer, but JSON data is left as a string?
-      // Related to https://github.com/nock/nock/issues/1542 ?
-    }
-
-    interceptor.markConsumed()
-
-    if (req.aborted) {
-      return
+    if (isJSON) {
+      defaultHeaders.push('Content-Type', 'application/json')
     }
 
     response.rawHeaders.push(
-      ...selectDefaultHeaders(
-        response.rawHeaders,
-        interceptor.scope._defaultReplyHeaders
-      )
+      ...selectDefaultHeaders(response.rawHeaders, defaultHeaders)
     )
 
     // Evaluate functional headers.
     common.forEachHeader(response.rawHeaders, (value, fieldName, i) => {
       if (typeof value === 'function') {
-        response.rawHeaders[i + 1] = value(req, response, responseBody)
+        response.rawHeaders[i + 1] = value(req, response, body)
       }
     })
 
     response.headers = common.headersArrayToObject(response.rawHeaders)
-
-    process.nextTick(() =>
-      respondUsingInterceptor({
-        responseBody,
-        responseBuffers,
-      })
-    )
   }
 
-  function respondUsingInterceptor({ responseBody, responseBuffers }) {
-    if (req.aborted) {
-      return
-    }
+  function continueWithResponseBody(rawBody) {
+    prepareResponseHeaders(rawBody)
+    const bodyAsStream = convertBodyToStream(rawBody)
+    bodyAsStream.pause()
+
+    // IncomingMessage extends Readable so we can't simply pipe.
+    bodyAsStream.on('data', function (chunk) {
+      response.push(chunk)
+    })
+    bodyAsStream.on('end', function () {
+      // https://nodejs.org/dist/latest-v10.x/docs/api/http.html#http_message_complete
+      response.complete = true
+      response.push(null)
+
+      interceptor.scope.emit('replied', req, interceptor)
+    })
+    bodyAsStream.on('error', function (err) {
+      response.emit('error', err)
+    })
+
+    const { delayBodyInMs, delayConnectionInMs } = interceptor
 
     function respond() {
-      if (req.aborted) {
+      if (common.isRequestDestroyed(req)) {
         return
       }
 
-      debug('emitting response')
+      // Even though we've had the response object for awhile at this point,
+      // we only attach it to the request immediately before the `response`
+      // event because, as in Node, it alters the error handling around aborts.
+      req.res = response
+      response.req = req
+
+      logger('emitting response')
       req.emit('response', response)
 
-      if (common.isStream(responseBody)) {
-        debug('resuming response stream')
-        responseBody.resume()
-      } else {
-        responseBuffers = responseBuffers || []
-        if (typeof responseBody !== 'undefined') {
-          debug('adding body to buffer list')
-          responseBuffers.push(responseBody)
-        }
-
-        // Stream the response chunks one at a time.
-        common.setImmediate(function emitChunk() {
-          const chunk = responseBuffers.shift()
-
-          if (chunk) {
-            debug('emitting response chunk')
-            response.push(chunk)
-            common.setImmediate(emitChunk)
-          } else {
-            debug('ending response stream')
-            response.push(null)
-            // https://nodejs.org/dist/latest-v10.x/docs/api/http.html#http_message_complete
-            response.complete = true
-            interceptor.scope.emit('replied', req, interceptor)
-          }
-        })
-      }
+      common.setTimeout(() => bodyAsStream.resume(), delayBodyInMs)
     }
 
-    if (interceptor.socketDelayInMs && interceptor.socketDelayInMs > 0) {
-      socket.applyDelay(interceptor.socketDelayInMs)
-    }
-
-    if (
-      interceptor.delayConnectionInMs &&
-      interceptor.delayConnectionInMs > 0
-    ) {
-      socket.applyDelay(interceptor.delayConnectionInMs)
-      common.setTimeout(respond, interceptor.delayConnectionInMs)
-    } else {
-      respond()
-    }
+    socket.applyDelay(delayConnectionInMs)
+    common.setTimeout(respond, delayConnectionInMs)
   }
 
-  start()
+  // Calling `start` immediately could take the request all the way to the connection delay
+  // during a single microtask execution. This setImmediate stalls the playback to ensure the
+  // correct events are emitted first ('socket', 'finish') and any aborts in the in the queue or
+  // called during a 'finish' listener can be called.
+  common.setImmediate(() => {
+    if (!common.isRequestDestroyed(req)) {
+      start()
+    }
+  })
 }
 
 module.exports = { playbackInterceptor }