Repository: mcollina/autocannon
Pull request #298 (2020-11-11 06:07:48)

feat: setupRequest/onResponse support for worker mode
- Added support for setupClient, setupRequest and onResponse when using workers
- Refactored code to reuse appropriately
- Added tests.


Patch:
diff --git a/README.md b/README.md
index 557ecc6..bfdee12 100644
--- a/README.md
+++ b/README.md
@@ -224,6 +224,47 @@ async function foo () {
 
 ```
 
+<a name="workers"></a>
+#### Workers
+
+```js
+'use strict'
+
+const autocannon = require('autocannon')
+
+autocannon({
+  url: 'http://localhost:3000',
+  connections: 10, //default
+  pipelining: 1, // default
+  duration: 10, // default
+  workers: 4
+}, console.log)
+```
+
+**NOTE:** When in workers mode, you need to pass in an absolute file path to all the options that accept a `function`. This is because a function passed in to the main process can not be cloned and passed to the worker. So instead, it needs a file that it can `require`. The options with this behaviour are shown in the below example
+
+```js
+'use strict'
+
+const autocannon = require('autocannon')
+
+autocannon({
+  // ...
+  workers: 4,
+  setupClient: '/full/path/to/setup-client.js',
+  requests: [
+    {
+      // ...
+      onResponse: '/full/path/to/on-response.js'
+    },
+    {
+      // ...
+      setupRequest: '/full/path/to/setup-request.js'
+    }
+  ]
+}, console.log)
+```
+
 ## API
 
 ### autocannon(opts[, cb])
@@ -245,7 +286,7 @@ Start autocannon against the given target.
     * `body`: A `String` or a `Buffer` containing the body of the request. Insert one or more randomly generated IDs into the body by including `[<id>]` where the randomly generated ID should be inserted (Must also set idReplacement to true). This can be useful in soak testing POST endpoints where one or more fields must be unique. Leave undefined for an empty body. _OPTIONAL_ default: `undefined`.
     * `form`: A `String` or an `Object` containing the multipart/form-data options or a path to the JSON file containing them
     * `headers`: An `Object` containing the headers of the request. _OPTIONAL_ default: `{}`.
-    * `setupClient`: A `Function` which will be passed the `Client` object for each connection to be made. This can be used to customise each individual connection headers and body using the API shown below. The changes you make to the client in this function will take precedence over the default `body` and `headers` you pass in here. There is an example of this in the samples folder. _OPTIONAL_ default: `function noop () {}`.
+    * `setupClient`: A `Function` which will be passed the `Client` object for each connection to be made. This can be used to customise each individual connection headers and body using the API shown below. The changes you make to the client in this function will take precedence over the default `body` and `headers` you pass in here. There is an example of this in the samples folder. _OPTIONAL_ default: `function noop () {}`. When using `workers`, you need to supply a file path that default exports a function instead (Check out [workers](#workers) section for more details).
     * `maxConnectionRequests`: A `Number` stating the max requests to make per connection. `amount` takes precedence if both are set. _OPTIONAL_
     * `maxOverallRequests`: A `Number` stating the max requests to make overall. Can't be less than `connections`. `maxConnectionRequests` takes precedence if both are set. _OPTIONAL_
     * `connectionRate`: A `Number` stating the rate of requests to make per second from each individual connection. No rate limiting by default. _OPTIONAL_
@@ -257,14 +298,15 @@ Start autocannon against the given target.
        * `headers`: When present, will override `opts.headers`. _OPTIONAL_
        * `method`: When present, will override `opts.method`. _OPTIONAL_
        * `path`: When present, will override `opts.path`. _OPTIONAL_
-       * `setupRequest`: A `Function` you may provide to mutate the raw `request` object, e.g. `request.method = 'GET'`. It takes `request` (Object) and `context` (Object) parameters, and must return the modified request. When it returns a falsey value, autocannon will restart from first request. _OPTIONAL_
-       * `onResponse`: A `Function` you may provide to process the received response. It takes `status` (Number), `body` (String) and `context` (Object) parameters. _OPTIONAL_
+       * `setupRequest`: A `Function` you may provide to mutate the raw `request` object, e.g. `request.method = 'GET'`. It takes `request` (Object) and `context` (Object) parameters, and must return the modified request. When it returns a falsey value, autocannon will restart from first request. When using `workers`, you need to supply a file path that default exports a function instead (Check out [workers](#workers) section for more details) _OPTIONAL_
+       * `onResponse`: A `Function` you may provide to process the received response. It takes `status` (Number), `body` (String) and `context` (Object) parameters. When using `workers`, you need to supply a file path that default exports a function instead (Check out [workers](#workers) section for more details) _OPTIONAL_
     * `har`: an `Object` of parsed [HAR](https://w3c.github.io/web-performance/specs/HAR/Overview.html) content. Autocannon will extra and use `entries.request`: `requests`, `method`, `form` and `body` options will be ignored. _NOTE_: you must ensure that entries are targeting the same domain as `url` option. _OPTIONAL_
     * `idReplacement`: A `Boolean` which enables the replacement of `[<id>]` tags within the request body with a randomly generated ID, allowing for unique fields to be sent with requests. Check out [an example of programmatic usage](./samples/using-id-replacement.js) can be found in the samples. _OPTIONAL_ default: `false`
     * `forever`: A `Boolean` which allows you to setup an instance of autocannon that restarts indefinitely after emiting results with the `done` event. Useful for efficiently restarting your instance. To stop running forever, you must cause a `SIGINT` or call the `.stop()` function on your instance. _OPTIONAL_ default: `false`
     * `servername`: A `String` identifying the server name for the SNI (Server Name Indication) TLS extension. _OPTIONAL_ default: Defaults to the hostname of the URL when it is not an IP address.
     * `excludeErrorStats`: A `Boolean` which allows you to disable tracking non 2xx code responses in latency and bytes per second calculations. _OPTIONAL_ default: `false`.
     * `expectBody`: A `String` representing the expected response body. Each request whose response body is not equal to `expectBody`is counted in `mismatches`. If enabled, mismatches count towards bailout. _OPTIONAL_
+    * `tlsOptions`: An `Object` that is passed into `tls.connect` call ([Full list of options](https://nodejs.org/api/tls.html#tls_tls_connect_port_host_options_callback)). Note: this only applies if your url is secure.
 * `cb`: The callback which is called on completion of a benchmark. Takes the following params. _OPTIONAL_.
     * `err`: If there was an error encountered with the run.
     * `results`: The results of the run.
diff --git a/autocannon.js b/autocannon.js
index e2a0abf..1199be5 100755
--- a/autocannon.js
+++ b/autocannon.js
@@ -13,10 +13,9 @@ const spawn = require('child_process').spawn
 const managePath = require('manage-path')
 const hasAsyncHooks = require('has-async-hooks')
 const help = fs.readFileSync(path.join(__dirname, 'help.txt'), 'utf8')
-const run = require('./lib/run')
 const printResult = require('./lib/printResult')
+const initJob = require('./lib/init')
 const track = require('./lib/progressTracker')
-const runTracker = require('./lib/runTracker')
 const { checkURL, ofURL } = require('./lib/url')
 const { parseHAR } = require('./lib/parseHAR')
 
@@ -25,7 +24,7 @@ if (typeof URL !== 'function') {
   process.exit(1)
 }
 
-module.exports = run
+module.exports = initJob
 module.exports.track = track
 
 module.exports.start = start
@@ -178,6 +177,10 @@ function parseArguments (argvs) {
     }
   }
 
+  // This is to distinguish down the line whether it is
+  // run via command-line or programatically
+  argv[Symbol.for('internal')] = true
+
   return argv
 }
 
@@ -199,7 +202,7 @@ function start (argv) {
         onPort: false,
         url: url
       })
-      runTracker(opts, () => {
+      initJob(opts, () => {
         proc.kill('SIGINT')
         server.close()
       })
@@ -220,7 +223,7 @@ function start (argv) {
       })
     })
   } else {
-    runTracker(argv)
+    initJob(argv)
   }
 }
 
diff --git a/lib/defaultOptions.js b/lib/defaultOptions.js
index 231d08e..10e5946 100644
--- a/lib/defaultOptions.js
+++ b/lib/defaultOptions.js
@@ -2,7 +2,6 @@
 
 const defaultOptions = {
   headers: {},
-  body: Buffer.alloc(0),
   method: 'GET',
   duration: 10,
   connections: 10,
diff --git a/lib/httpClient.js b/lib/httpClient.js
index 0e7ef80..0250c77 100644
--- a/lib/httpClient.js
+++ b/lib/httpClient.js
@@ -7,6 +7,7 @@ const tls = require('tls')
 const retimer = require('retimer')
 const HTTPParser = require('http-parser-js').HTTPParser
 const RequestIterator = require('./requestIterator')
+const noop = require('./noop')
 const clone = require('clone')
 
 function Client (opts) {
@@ -21,6 +22,7 @@ function Client (opts) {
   const pipelining = this.opts.pipelining
   this.opts.port = this.opts.port || 80
   this.opts.expectBody = this.opts.expectBody || null
+  this.opts.tlsOptions = this.opts.tlsOptions || {}
   this.timeout = (this.opts.timeout || 10) * 1000
   this.ipc = !!this.opts.socketPath
   this.secure = this.opts.protocol === 'https:'
@@ -131,9 +133,9 @@ Client.prototype._connect = function () {
     }
 
     if (this.ipc) {
-      this.conn = tls.connect(this.opts.socketPath, { rejectUnauthorized: false })
+      this.conn = tls.connect(this.opts.socketPath, { ...this.opts.tlsOptions, rejectUnauthorized: false })
     } else {
-      this.conn = tls.connect(this.opts.port, this.opts.hostname, { rejectUnauthorized: false, servername })
+      this.conn = tls.connect(this.opts.port, this.opts.hostname, { ...this.opts.tlsOptions, rejectUnauthorized: false, servername })
     }
   } else {
     if (this.ipc) {
@@ -242,6 +244,4 @@ Client.prototype._okayToUpdateCheck = function () {
   }
 }
 
-function noop () {}
-
 module.exports = Client
diff --git a/lib/init.js b/lib/init.js
new file mode 100644
index 0000000..59e4d0d
--- /dev/null
+++ b/lib/init.js
@@ -0,0 +1,69 @@
+'use strict'
+
+const EE = require('events').EventEmitter
+const { isMainThread } = require('./worker_threads')
+
+const initWorkers = require('./manager')
+const validateOpts = require('./validate')
+const noop = require('./noop')
+const runTracker = require('./runTracker')
+const track = require('./progressTracker')
+
+function init (opts, cb) {
+  return _init(opts, null, cb)
+}
+
+function _init (opts, tracker, cb) {
+  const cbPassedIn = (typeof cb === 'function')
+  cb = cb || noop
+
+  tracker = tracker || new EE()
+
+  if (!cbPassedIn && !opts.forever) {
+    const promise = new Promise((resolve, reject) => {
+      _init(opts, tracker, (err, results) => {
+        if (err) return reject(err)
+        resolve(results)
+      })
+    })
+    tracker.then = promise.then.bind(promise)
+    tracker.catch = promise.catch.bind(promise)
+    return tracker
+  }
+
+  opts = validateOpts(opts, cbPassedIn)
+
+  function _cb (err, result) {
+    if (err) {
+      return cbPassedIn ? cb(err) : setImmediate(() => tracker.emit('error', err))
+    }
+
+    setImmediate(() => tracker.emit('done', result))
+    cb(null, result)
+
+    if (!err && opts.json) {
+      console.log(JSON.stringify(result))
+    }
+  }
+
+  if (opts instanceof Error) {
+    _cb(opts)
+    return tracker
+  }
+
+  tracker.opts = opts
+
+  if (opts.workers && isMainThread) {
+    initWorkers(opts, tracker, _cb)
+  } else {
+    runTracker(opts, tracker, _cb)
+  }
+
+  // if not running via command-line and
+  // not rendering json, or if std isn't a tty, track progress
+  if (opts[Symbol.for('internal')] && (!opts.json || !process.stdout.isTTY)) track(tracker, opts)
+
+  return tracker
+}
+
+module.exports = init
diff --git a/lib/manager.js b/lib/manager.js
index f05993b..92a9986 100644
--- a/lib/manager.js
+++ b/lib/manager.js
@@ -1,12 +1,15 @@
 'use strict'
 
 const path = require('path')
+const EE = require('events').EventEmitter
 const aggregateResult = require('./aggregateResult')
 const { getHistograms } = require('./histUtil')
 const ora = require('ora')
 const { Worker } = require('./worker_threads')
 
-function initWorkers (opts, cb) {
+function initWorkers (opts, tracker, cb) {
+  tracker = tracker || new EE()
+
   const workers = []
   const results = []
   const numWorkers = +opts.workers
@@ -54,12 +57,12 @@ function initWorkers (opts, cb) {
     cb(null, result)
     reset()
 
-    setImmediate(() => {
-      if (opts.forever && restart) {
+    if (opts.forever && restart) {
+      setImmediate(() => {
         init()
         startAll()
-      }
-    })
+      })
+    }
   }
 
   function stopAll () {
@@ -77,8 +80,8 @@ function initWorkers (opts, cb) {
 
   const workerOpts = {
     ...opts,
-    amount: Math.max(Math.floor(opts.amount / numWorkers), 1),
-    connections: Math.max(Math.floor(opts.connections / numWorkers), 1)
+    ...(opts.amount ? { amount: Math.max(Math.floor(opts.amount / numWorkers), 1) } : undefined),
+    ...(opts.connections ? { connections: Math.max(Math.floor(opts.connections / numWorkers), 1) } : undefined)
   }
   workerOpts.a = workerOpts.amount
   workerOpts.c = workerOpts.connections
@@ -88,7 +91,7 @@ function initWorkers (opts, cb) {
       const worker = new Worker(path.resolve(__dirname, './worker.js'), { workerData: { opts: workerOpts } })
 
       worker.on('message', (msg) => {
-        const { cmd, data } = msg
+        const { cmd, data, error } = msg
 
         if (cmd === 'RESULT') {
           results.push(data)
@@ -108,6 +111,8 @@ function initWorkers (opts, cb) {
         } else if (cmd === 'RESET_HIST') {
           const { name } = data
           histograms[name].reset()
+        } else if (cmd === 'ERROR') {
+          tracker.emit('error', error)
         }
       })
 
@@ -124,12 +129,12 @@ function initWorkers (opts, cb) {
   init()
   startAll()
 
-  return {
-    stop: () => {
-      restart = false
-      stopAll()
-    }
+  tracker.stop = () => {
+    restart = false
+    stopAll()
   }
+
+  return tracker
 }
 
 module.exports = initWorkers
diff --git a/lib/noop.js b/lib/noop.js
new file mode 100644
index 0000000..9a8b060
--- /dev/null
+++ b/lib/noop.js
@@ -0,0 +1,2 @@
+/* istanbul ignore next */
+module.exports = function noop () {}
diff --git a/lib/progressTracker.js b/lib/progressTracker.js
index 32570ef..a93951b 100644
--- a/lib/progressTracker.js
+++ b/lib/progressTracker.js
@@ -57,16 +57,14 @@ function track (instance, opts) {
     }
   })
 
-  if (!opts.workers) {
-    instance.on('done', (result) => {
-      // the code below this `if` just renders the results table...
-      // if the user doesn't want to render the table, we can just return early
-      if (opts.renderResultsTable === false) return
-
-      const tableStrResult = printResult(result, opts)
-      opts.outputStream.write(tableStrResult)
-    })
-  }
+  instance.on('done', (result) => {
+    // the code below this `if` just renders the results table...
+    // if the user doesn't want to render the table, we can just return early
+    if (opts.renderResultsTable === false) return
+
+    const tableStrResult = printResult(result, opts)
+    opts.outputStream.write(tableStrResult)
+  })
 
   function addListener () {
     // add listeners for progress bar to instance here so they aren't
diff --git a/lib/run.js b/lib/run.js
index 2dd90fb..c8f93e7 100644
--- a/lib/run.js
+++ b/lib/run.js
@@ -1,85 +1,22 @@
 'use strict'
 
-const EE = require('events').EventEmitter
 const URL = require('url')
-const timestring = require('timestring')
 const reInterval = require('reinterval')
+const EE = require('events').EventEmitter
 const Client = require('./httpClient')
-const DefaultOptions = require('./defaultOptions')
-const multipart = require('./multipart')
 const { isMainThread } = require('./worker_threads')
-const { ofURL, checkURL } = require('./url')
-const { parseHAR } = require('./parseHAR')
-const initWorkers = require('./manager')
+const { ofURL } = require('./url')
 const aggregateResult = require('./aggregateResult')
-const printResult = require('./printResult')
 const { getHistograms, encodeHist } = require('./histUtil')
 
-function handleFinish (result, opts, cb) {
-  if (cb) cb(null, result)
-  if (opts.json) {
-    console.log(JSON.stringify(result))
-  }
-
-  // the code below this `if` just renders the results table...
-  // if the user doesn't want to render the table, we can just return early
-  if (!isMainThread || opts.renderResultsTable === false) return
-
-  const tableStrResult = printResult(result, opts);
-  (opts.outputStream ? opts.outputStream : process.stderr).write(tableStrResult)
+const defaults = {
+  harRequests: new Map()
 }
 
-function validateOpts (opts, cb) {
-  if (opts.workers) {
-    if (opts.setupClient) {
-      cb(new Error('Can not use setupClient option while using workers'))
-      return
-    }
-
-    if (opts.requests && opts.requests.some(r => r.setupRequest || r.onResponse)) {
-      cb(new Error('Can not use setupRequest/onResponse options while using workers'))
-    }
-  }
-
-  return true
-}
-
-function run (opts, cb) {
-  return _run(opts, cb)
-}
-
-function _run (opts, cb, tracker) {
-  const cbPassedIn = (typeof cb === 'function')
-
-  cb = cb || noop
-
+function run (opts, tracker, cb) {
+  opts = Object.assign({}, defaults, opts)
   tracker = tracker || new EE()
 
-  if (!cbPassedIn && !opts.forever) {
-    const promise = new Promise((resolve, reject) => {
-      _run(opts, function (err, results) {
-        if (err) return reject(err)
-        resolve(results)
-      }, tracker)
-    })
-    tracker.then = promise.then.bind(promise)
-    tracker.catch = promise.catch.bind(promise)
-    return tracker
-  }
-
-  if (opts.workers && isMainThread) {
-    if (!validateOpts(opts, cb)) return
-
-    return initWorkers(opts, (err, result) => {
-      if (err) {
-        console.log(err)
-        process.exit(1)
-      }
-
-      handleFinish(result, opts, cb)
-    })
-  }
-
   const histograms = getHistograms(opts.histograms)
   const { latencies, requests, throughput } = histograms
 
@@ -91,18 +28,6 @@ function _run (opts, cb, tracker) {
     0 // 5xx
   ]
 
-  if (opts && opts.form) {
-    opts.method = opts.method || 'POST'
-  }
-  opts = Object.assign({}, DefaultOptions, opts)
-
-  // do error checking, if error, return
-  if (checkOptsForErrors()) return tracker
-
-  // set tracker.opts here, so throwing over invalid opts and setting defaults etc.
-  // is done
-  tracker.opts = opts
-
   if (opts.overallRate && (opts.overallRate < opts.connections)) opts.connections = opts.overallRate
 
   let counter = 0
@@ -120,32 +45,12 @@ function _run (opts, cb, tracker) {
   let numRunning = opts.connections
   let startTime = Date.now()
   const includeErrorStats = !opts.excludeErrorStats
-  let form
-
-  if (opts.form) {
-    try {
-      form = multipart(opts.form)
-    } catch (error) {
-      errorCb(error)
-      return tracker
-    }
-  }
 
   opts.url = ofURL(opts.url).map((url) => {
     if (url.indexOf('http') !== 0) return 'http://' + url
     return url
   })
 
-  let harRequests = new Map()
-  if (opts.har) {
-    try {
-      harRequests = parseHAR(opts.har)
-    } catch (error) {
-      errorCb(error)
-      return tracker
-    }
-  }
-
   const urls = ofURL(opts.url, true).map(url => {
     if (url.indexOf('http') !== 0) url = 'http://' + url
     url = URL.parse(url) // eslint-disable-line node/no-deprecated-api
@@ -154,13 +59,13 @@ function _run (opts, cb, tracker) {
     // performs the right HTTP requests
     url.pipelining = opts.pipelining
     url.method = opts.method
-    url.body = form ? form.getBuffer() : opts.body
-    url.headers = form ? Object.assign({}, opts.headers, form.getHeaders()) : opts.headers
+    url.body = opts.form ? opts.form.getBuffer() : opts.body
+    url.headers = opts.form ? Object.assign({}, opts.headers, opts.form.getHeaders()) : opts.headers
     url.setupClient = opts.setupClient
     url.timeout = opts.timeout
     url.origin = `${url.protocol}//${url.host}`
     // only keep requests for that origin, or default to requests from options
-    url.requests = harRequests.get(url.origin) || opts.requests
+    url.requests = opts.harRequests.get(url.origin) || opts.requests
     url.reconnectRate = opts.reconnectRate
     url.responseMax = amount || opts.maxConnectionRequests || opts.maxOverallRequests
     url.rate = opts.connectionRate || opts.overallRate
@@ -181,11 +86,6 @@ function _run (opts, cb, tracker) {
     }, opts.duration * 1000)
   }
 
-  tracker.stop = () => {
-    stop = true
-    restart = false
-  }
-
   const interval = reInterval(tickInterval, 1000)
 
   // put the start emit in a setImmediate so trackers can be added, etc.
@@ -225,9 +125,11 @@ function _run (opts, cb, tracker) {
 
       const resultObj = isMainThread ? aggregateResult(result, opts, histograms) : result
 
-      tracker.emit('done', resultObj)
-
-      if (!opts.forever) {
+      if (opts.forever) {
+        // we don't call callback when in forever mode, so this is the
+        // only place we could notify user when each round finishes
+        tracker.emit('done', resultObj)
+      } else {
         latencies.destroy()
         requests.destroy()
         throughput.destroy()
@@ -284,6 +186,10 @@ function _run (opts, cb, tracker) {
         url.rate = distributeNums(opts.overallRate, i)
       }
 
+      if (opts.tlsOptions) {
+        url.tlsOptions = opts.tlsOptions
+      }
+
       const client = new Client(url)
       client.on('response', onResponse)
       client.on('connError', onError)
@@ -348,91 +254,12 @@ function _run (opts, cb, tracker) {
     }
   }
 
-  function errorCb (error) {
-    if (cbPassedIn) {
-      cb(error)
-    } else {
-      // wrapped in setImmediate so any error event handlers that are added to
-      // the tracker can be added before being emitted
-      setImmediate(() => { tracker.emit('error', error) })
-    }
+  tracker.stop = () => {
+    stop = true
+    restart = false
   }
 
-  // will return true if error with opts entered
-  function checkOptsForErrors () {
-    if (!checkURL(opts.url) && !opts.socketPath) {
-      errorCb(new Error('url or socketPath option required'))
-      return true
-    }
-
-    if (typeof opts.duration === 'string') {
-      if (/[a-zA-Z]/.exec(opts.duration)) opts.duration = timestring(opts.duration)
-      else opts.duration = Number(opts.duration.trim())
-    }
-
-    if (typeof opts.duration === 'number') {
-      if (lessThanZeroError(opts.duration, 'duration')) return true
-    } else {
-      errorCb(new Error('duration entered was in an invalid format'))
-      return true
-    }
-
-    if (opts.expectBody && opts.requests !== DefaultOptions.requests) {
-      errorCb(new Error('expectBody cannot be used in conjunction with requests'))
-      return true
-    }
-
-    if (lessThanOneError(opts.connections, 'connections')) return true
-    if (lessThanOneError(opts.pipelining, 'pipelining factor')) return true
-    if (greaterThanZeroError(opts.timeout, 'timeout')) return true
-    if (opts.bailout && lessThanOneError(opts.bailout, 'bailout threshold')) return true
-    if (opts.connectionRate && lessThanOneError(opts.connectionRate, 'connectionRate')) return true
-    if (opts.overallRate && lessThanOneError(opts.overallRate, 'bailout overallRate')) return true
-    if (opts.amount && lessThanOneError(opts.amount, 'amount')) return true
-    if (opts.maxConnectionRequests && lessThanOneError(opts.maxConnectionRequests, 'maxConnectionRequests')) return true
-    if (opts.maxOverallRequests && lessThanOneError(opts.maxOverallRequests, 'maxOverallRequests')) return true
-
-    if (opts.ignoreCoordinatedOmission && !opts.connectionRate && !opts.overallRate) {
-      errorCb(new Error('ignoreCoordinatedOmission makes no sense without connectionRate or overallRate'))
-      return true
-    }
-
-    if (opts.forever && cbPassedIn) {
-      errorCb(new Error('should not use the callback parameter when the `forever` option is set to true. Use the `done` event on this event emitter'))
-      return true
-    }
-
-    function lessThanZeroError (x, label) {
-      if (x < 0) {
-        errorCb(new Error(`${label} can not be less than 0`))
-        return true
-      }
-      return false
-    }
-
-    function lessThanOneError (x, label) {
-      if (x < 1) {
-        errorCb(new Error(`${label} can not be less than 1`))
-        return true
-      }
-      return false
-    }
-
-    function greaterThanZeroError (x, label) {
-      if (x <= 0) {
-        errorCb(new Error(`${label} must be greater than 0`))
-        return true
-      }
-      return false
-    }
-
-    return false
-  } // checkOptsForErrors
-
   return tracker
 } // run
 
-/* istanbul ignore next */
-function noop () {}
-
 module.exports = run
diff --git a/lib/runTracker.js b/lib/runTracker.js
index dfb5ca1..5aaeff5 100644
--- a/lib/runTracker.js
+++ b/lib/runTracker.js
@@ -1,33 +1,20 @@
 'use strict'
 
+const EE = require('events').EventEmitter
 const run = require('./run')
-const track = require('./progressTracker')
+const noop = require('./noop')
 
-function runTracker (argv, ondone) {
-  const tracker = run(argv)
+function runTracker (opts, tracker, cb) {
+  cb = cb || noop
+  tracker = tracker || new EE()
 
-  // While using workers, tracking is handled in manager.js
-  if (argv.workers) return
-
-  tracker.on('done', (result) => {
-    if (ondone) ondone()
-    if (argv.json) {
-      console.log(JSON.stringify(result))
-    }
-  })
-
-  tracker.on('error', (err) => {
-    if (err) {
-      throw err
-    }
-  })
-
-  // if not rendering json, or if std isn't a tty, track progress
-  if (!argv.json || !process.stdout.isTTY) track(tracker, argv)
+  run(opts, tracker, cb)
 
   process.once('SIGINT', () => {
     tracker.stop()
   })
+
+  return tracker
 }
 
 module.exports = runTracker
diff --git a/lib/validate.js b/lib/validate.js
new file mode 100644
index 0000000..386d363
--- /dev/null
+++ b/lib/validate.js
@@ -0,0 +1,136 @@
+'use strict'
+
+const defaultOptions = require('./defaultOptions')
+const timestring = require('timestring')
+const { checkURL } = require('./url')
+const multipart = require('./multipart')
+const { parseHAR } = require('./parseHAR')
+
+const isValidFn = (opt) => (!opt || typeof opt === 'function' || typeof opt === 'string')
+
+const lessThanOneError = (label) => new Error(`${label} can not be less than 1`)
+const greaterThanZeroError = (label) => new Error(`${label} must be greater than 0`)
+const minIfPresent = (val, min) => val && val < min
+
+function safeRequire (path) {
+  if (typeof path === 'string') {
+    try {
+      return require(path)
+    } catch (err) {}
+  }
+
+  return path
+}
+
+function defaultOpts (opts) {
+  const setupClient = opts.workers ? opts.setupClient : safeRequire(opts.setupClient)
+
+  const requests = opts.requests
+    ? opts.requests.map((r) => {
+        const setupRequest = opts.workers ? r.setupRequest : safeRequire(r.setupRequest)
+        const onResponse = opts.workers ? r.onResponse : safeRequire(r.onResponse)
+
+        return {
+          ...r,
+          ...(setupRequest ? { setupRequest } : undefined),
+          ...(onResponse ? { onResponse } : undefined)
+        }
+      })
+    : undefined
+
+  return {
+    ...defaultOptions,
+    ...opts,
+    ...(setupClient ? { setupClient } : undefined),
+    ...(requests ? { requests } : undefined)
+  }
+}
+
+module.exports = function validateOpts (opts, cbPassedIn) {
+  // these need to be validated before defaulting
+  if (minIfPresent(opts.bailout, 1)) return lessThanOneError('bailout threshold')
+  if (minIfPresent(opts.connectionRate, 1)) return lessThanOneError('connectionRate')
+  if (minIfPresent(opts.overallRate, 1)) return lessThanOneError('bailout overallRate')
+  if (minIfPresent(opts.amount, 1)) return lessThanOneError('amount')
+  if (minIfPresent(opts.maxConnectionRequests, 1)) return lessThanOneError('maxConnectionRequests')
+  if (minIfPresent(opts.maxOverallRequests, 1)) return lessThanOneError('maxOverallRequests')
+
+  if (opts.form) {
+    opts.method = opts.method || 'POST'
+  }
+
+  // fill in defaults after
+  opts = defaultOpts(opts)
+
+  if (opts.requests) {
+    if (opts.requests.some(r => !isValidFn(r.setupRequest))) {
+      return new Error('Invalid option setupRequest, please provide a function (or file path when in workers mode)')
+    }
+
+    if (opts.requests.some(r => !isValidFn(r.onResponse))) {
+      return new Error('Invalid option onResponse, please provide a function (or file path when in workers mode)')
+    }
+  }
+
+  if (!isValidFn(opts.setupClient)) {
+    return new Error('Invalid option setupClient, please provide a function (or file path when in workers mode)')
+  }
+
+  if (!checkURL(opts.url) && !opts.socketPath) {
+    return new Error('url or socketPath option required')
+  }
+
+  if (typeof opts.duration === 'string') {
+    if (/[a-zA-Z]/.exec(opts.duration)) opts.duration = timestring(opts.duration)
+    else opts.duration = Number(opts.duration.trim())
+  }
+
+  if (typeof opts.duration !== 'number') {
+    return new Error('duration entered was in an invalid format')
+  }
+
+  if (opts.duration < 0) {
+    return new Error('duration can not be less than 0')
+  }
+
+  if (opts.expectBody && opts.requests !== defaultOptions.requests) {
+    return new Error('expectBody cannot be used in conjunction with requests')
+  }
+
+  if (opts.form) {
+    try {
+      // Parse multipart upfront to make sure there's no errors
+      const data = multipart(opts.form)
+      opts.form = opts.workers ? opts.form : data // but use parsed data only if not in workers mode
+    } catch (error) {
+      return error
+    }
+  }
+
+  opts.harRequests = new Map()
+  if (opts.har) {
+    try {
+      opts.harRequests = parseHAR(opts.har)
+    } catch (error) {
+      return error
+    }
+  }
+
+  if (opts.connections < 1) return lessThanOneError('connections')
+  if (opts.pipelining < 1) return lessThanOneError('pipelining factor')
+  if (opts.timeout < 1) return greaterThanZeroError('timeout')
+
+  if (opts.ignoreCoordinatedOmission && !opts.connectionRate && !opts.overallRate) {
+    return new Error('ignoreCoordinatedOmission makes no sense without connectionRate or overallRate')
+  }
+
+  if (opts.forever && cbPassedIn) {
+    return new Error('should not use the callback parameter when the `forever` option is set to true. Use the `done` event on this event emitter')
+  }
+
+  if (opts.forever && opts.workers) {
+    return new Error('Using `forever` option isn\'t currently supported with workers')
+  }
+
+  return opts
+}
diff --git a/lib/worker.js b/lib/worker.js
index 04b0cff..9c8953c 100644
--- a/lib/worker.js
+++ b/lib/worker.js
@@ -1,6 +1,7 @@
 'use strict'
 
 const { isMainThread, parentPort, workerData } = require('worker_threads')
+const multipart = require('./multipart')
 const run = require('./run')
 
 const createHist = (name) => ({
@@ -24,30 +25,31 @@ const resetHist = (name) => {
   })
 }
 
-function runTracker (argv, ondone) {
+function runTracker (opts, cb) {
   const tracker = run({
-    ...argv,
+    ...opts,
+    ...(opts.form ? { form: multipart(opts.form) } : undefined),
+    ...(opts.setupClient ? { setupClient: require(opts.setupClient) } : undefined),
+    requests: opts.requests
+      ? opts.requests.map(r => ({
+          ...r,
+          ...(r.setupRequest ? { setupRequest: require(r.setupRequest) } : undefined),
+          ...(r.onResponse ? { onResponse: require(r.onResponse) } : undefined)
+        }))
+      : undefined,
     histograms: {
       requests: createHist('requests'),
       throughput: createHist('throughput')
     }
-  })
-
-  tracker.on('done', (result) => {
-    if (ondone) ondone(result)
-  })
+  }, null, cb)
 
   tracker.on('tick', (data) => {
     parentPort.postMessage({ cmd: 'TICK', data })
   })
 
-  tracker.on('error', (err) => {
-    if (err) {
-      throw err
-    }
-  })
-
-  return tracker
+  return {
+    stop: tracker.stop
+  }
 }
 
 if (!isMainThread) {
@@ -58,8 +60,8 @@ if (!isMainThread) {
     const { cmd } = msg
 
     if (cmd === 'START') {
-      tracker = runTracker(opts, (data) => {
-        parentPort.postMessage({ cmd: 'RESULT', data })
+      tracker = runTracker(opts, (error, data) => {
+        parentPort.postMessage({ cmd: error ? 'ERROR' : 'RESULT', error, data })
         parentPort.close()
       })
     } else if (cmd === 'STOP') {
diff --git a/lib/worker_threads.js b/lib/worker_threads.js
index 0d14b3c..0263ec4 100644
--- a/lib/worker_threads.js
+++ b/lib/worker_threads.js
@@ -7,7 +7,7 @@ try {
 } catch (err) {
   // we don't need the error but can't have catch block
   // without err as node 8 doesn't support that
-  if (err) console.log(err)
+  if (err) console.log('worker_threads is not available')
 
   workerThreads = {
     isMainThread: true
diff --git a/samples/helpers/on-response.js b/samples/helpers/on-response.js
new file mode 100644
index 0000000..573897f
--- /dev/null
+++ b/samples/helpers/on-response.js
@@ -0,0 +1,7 @@
+'use strict'
+
+module.exports = (status, body, context) => {
+  if (status === 200) {
+    context.user = JSON.parse(body)
+  } // on error, you may abort the benchmark
+}
diff --git a/samples/helpers/setup-request.js b/samples/helpers/setup-request.js
new file mode 100644
index 0000000..bc1af20
--- /dev/null
+++ b/samples/helpers/setup-request.js
@@ -0,0 +1,10 @@
+'use strict'
+
+module.exports = (req, context) => ({
+  ...req,
+  path: `/user/${context.user.id}`,
+  body: JSON.stringify({
+    ...context.user,
+    lastName: 'Doe'
+  })
+})
diff --git a/samples/request-context-workers.js b/samples/request-context-workers.js
new file mode 100644
index 0000000..7dea271
--- /dev/null
+++ b/samples/request-context-workers.js
@@ -0,0 +1,45 @@
+'use strict'
+
+const http = require('http')
+const path = require('path')
+const autocannon = require('../autocannon')
+
+const server = http.createServer(handle)
+
+server.listen(0, startBench)
+
+function handle (req, res) {
+  const body = []
+  // this route handler simply returns whatever it gets from response
+  req
+    .on('data', chunk => body.push(chunk))
+    .on('end', () => res.end(Buffer.concat(body)))
+}
+
+function startBench () {
+  const url = 'http://localhost:' + server.address().port
+
+  autocannon({
+    url: url,
+    duration: 2,
+    workers: 2,
+    requests: [
+      {
+        // let's create a new user
+        method: 'POST',
+        path: '/users',
+        body: JSON.stringify({ firstName: 'Jane', id: 10 }),
+        onResponse: path.join(__dirname, 'helpers', 'on-response')
+      },
+      {
+        // now we'll give them a last name
+        method: 'PUT',
+        setupRequest: path.join(__dirname, 'helpers', 'setup-request')
+      }
+    ]
+  }, finishedBench)
+
+  function finishedBench (err, res) {
+    console.log('finished bench', err, res)
+  }
+}
diff --git a/samples/track-run-workers.js b/samples/track-run-workers.js
index e1df3e7..8800c59 100644
--- a/samples/track-run-workers.js
+++ b/samples/track-run-workers.js
@@ -19,6 +19,8 @@ function startBench () {
     workers: 2
   }, finishedBench)
 
+  autocannon.track(instance)
+
   // this is used to kill the instance on CTRL-C
   process.once('SIGINT', () => {
     instance.stop()


Test patch:
diff --git a/test/cli.test.js b/test/cli.test.js
index 0f4ee76..a6f7166 100644
--- a/test/cli.test.js
+++ b/test/cli.test.js
@@ -6,10 +6,8 @@ const path = require('path')
 const fs = require('fs')
 const os = require('os')
 const childProcess = require('child_process')
-const semver = require('semver')
 const helper = require('./helper')
-
-const hasWorkerSupport = semver.gte(process.versions.node, '11.7.0')
+const hasWorkerSupport = require('./utils/has-worker-support')
 
 test('should run benchmark against server', (t) => {
   const lines = [
diff --git a/test/forever.test.js b/test/forever.test.js
index e24f707..e6f487d 100644
--- a/test/forever.test.js
+++ b/test/forever.test.js
@@ -1,14 +1,14 @@
 'use strict'
 
 const test = require('tap').test
-const run = require('../lib/run')
+const initJob = require('../lib/init')
 const helper = require('./helper')
 const server = helper.startServer()
 
 test('running with forever set to true and passing in a callback should cause an error to be returned in the callback', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     forever: true
   }, (err, res) => {
@@ -22,7 +22,7 @@ test('run forever should run until .stop() is called', (t) => {
   t.plan(3)
   let numRuns = 0
 
-  const instance = run({
+  const instance = initJob({
     url: `http://localhost:${server.address().port}`,
     duration: 0.5,
     forever: true
diff --git a/test/helper.js b/test/helper.js
index 6a8bbd4..c691350 100644
--- a/test/helper.js
+++ b/test/helper.js
@@ -107,7 +107,7 @@ function startHttpsServer (opts = {}) {
   return server
 }
 
-// this server will echo the SNI Server Name in a HTTP header
+// this server will echo the SNI Server Name and emailAddress from the client certificate in a HTTP header
 function startTlsServer () {
   const key = fs.readFileSync(path.join(__dirname, '/key.pem'))
   const cert = fs.readFileSync(path.join(__dirname, '/cert.pem'))
@@ -116,7 +116,9 @@ function startTlsServer () {
   const options = {
     key,
     cert,
-    passphrase
+    passphrase,
+    requestCert: true,
+    rejectUnauthorized: false
   }
 
   const server = tls.createServer(options, handle)
@@ -125,9 +127,16 @@ function startTlsServer () {
 
   function handle (socket) {
     const servername = socket.servername || ''
+    const certificate = socket.getPeerCertificate()
+    const email = (certificate && certificate.subject && certificate.subject.emailAddress) || ''
     socket.on('data', function (data) {
       // Assume this is a http get request and send back the servername in an otherwise empty reponse.
-      socket.write('HTTP/1.1 200 OK\nX-servername: ' + servername + '\nContent-Length: 0\n\n')
+      socket.write('HTTP/1.1 200 OK\n')
+      socket.write('X-servername: ' + servername + '\n')
+      if (email) {
+        socket.write('X-email: ' + email + '\n')
+      }
+      socket.write('Content-Length: 0\n\n')
       socket.setEncoding('utf8')
       socket.pipe(socket)
     })
diff --git a/test/httpClient.test.js b/test/httpClient.test.js
index 60873f6..14e4bc8 100644
--- a/test/httpClient.test.js
+++ b/test/httpClient.test.js
@@ -12,6 +12,7 @@ const httpsServer = helper.startHttpsServer()
 const tlsServer = helper.startTlsServer()
 const trailerServer = helper.startTrailerServer()
 const bl = require('bl')
+const fs = require('fs')
 
 const makeResponseFromBody = ({ server, method, body, headers = {} }) => {
   const sentHeaders = {
@@ -170,6 +171,28 @@ test('client ignores falsy SNI servername', (t) => {
   })
 })
 
+test('client passes through tlsOptions to connect', (t) => {
+  t.plan(4)
+
+  var opts = tlsServer.address()
+  opts.protocol = 'https:'
+  opts.tlsOptions = {
+    key: fs.readFileSync(path.join(__dirname, '/key.pem')),
+    cert: fs.readFileSync(path.join(__dirname, '/cert.pem')),
+    passphrase: 'test'
+  }
+  const client = new Client(opts)
+  let count = 0
+
+  client.on('headers', (response) => {
+    t.equal(response.statusCode, 200, 'status code matches')
+    t.deepEqual(response.headers, ['X-servername', '', 'X-email', 'tes@test.com', 'Content-Length', '0'])
+    if (count++ > 0) {
+      client.destroy()
+    }
+  })
+})
+
 test('http client automatically reconnects', (t) => {
   t.plan(4)
 
diff --git a/test/run.test.js b/test/run.test.js
index 533f2fe..a38b0e6 100644
--- a/test/run.test.js
+++ b/test/run.test.js
@@ -3,13 +3,13 @@
 const os = require('os')
 const path = require('path')
 const test = require('tap').test
-const run = require('../lib/run')
+const initJob = require('../lib/init')
 const defaultOptions = require('../lib/defaultOptions')
 const helper = require('./helper')
 const server = helper.startServer()
 
-test('run', (t) => {
-  run({
+test('init', (t) => {
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     duration: 2,
@@ -75,7 +75,7 @@ test('run', (t) => {
 })
 
 test('tracker.stop()', (t) => {
-  const tracker = run({
+  const tracker = initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     duration: 2
@@ -145,7 +145,7 @@ test('tracker.stop()', (t) => {
 })
 
 test('requests.min should be 0 when there are no successful requests', (t) => {
-  run({
+  initJob({
     url: 'nonexistent',
     connections: 1,
     amount: 1
@@ -159,7 +159,7 @@ test('requests.min should be 0 when there are no successful requests', (t) => {
 test('run should callback with an error with an invalid connections factor', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: -1
   }, function (err, result) {
@@ -172,7 +172,7 @@ test('run should callback with an error with an invalid connections factor', (t)
 test('run should callback with an error with an invalid pipelining factor', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     pipelining: -1
   }, function (err, result) {
@@ -185,7 +185,7 @@ test('run should callback with an error with an invalid pipelining factor', (t)
 test('run should callback with an error with an invalid bailout', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     bailout: -1
   }, function (err, result) {
@@ -198,7 +198,7 @@ test('run should callback with an error with an invalid bailout', (t) => {
 test('run should callback with an error with an invalid duration', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     duration: -1
   }, function (err, result) {
@@ -211,7 +211,7 @@ test('run should callback with an error with an invalid duration', (t) => {
 test('run should callback with an error when no url is passed in', (t) => {
   t.plan(2)
 
-  run({}, function (err, result) {
+  initJob({}, function (err, result) {
     t.ok(err, 'no url should cause an error')
     t.notOk(result, 'results should not exist')
     t.end()
@@ -221,7 +221,7 @@ test('run should callback with an error when no url is passed in', (t) => {
 test('run should callback with an error after a bailout', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:4', // 4 = first unassigned port: https://en.wikipedia.org/wiki/List_of_TCP_and_UDP_port_numbers
     bailout: 1
   }, function (err, result) {
@@ -234,7 +234,7 @@ test('run should callback with an error after a bailout', (t) => {
 test('run should callback with an error using expectBody and requests', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     requests: [{ body: 'something' }],
     expectBody: 'hello'
@@ -248,7 +248,7 @@ test('run should callback with an error using expectBody and requests', (t) => {
 test('run should allow users to enter timestrings to be used for duration', (t) => {
   t.plan(3)
 
-  const instance = run({
+  const instance = initJob({
     url: 'http://localhost:' + server.address().port,
     duration: '10m'
   }, function (err, result) {
@@ -267,7 +267,7 @@ test('run should allow users to enter timestrings to be used for duration', (t)
 test('run should recognise valid urls without http at the start', (t) => {
   t.plan(3)
 
-  run({
+  initJob({
     url: 'localhost:' + server.address().port,
     duration: 1
   }, (err, res) => {
@@ -281,7 +281,7 @@ test('run should recognise valid urls without http at the start', (t) => {
 test('run should produce count of mismatches with expectBody set', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     expectBody: 'body will not be this',
     maxOverallRequests: 10,
@@ -299,7 +299,7 @@ test('run should produce 0 mismatches with expectBody set and matches', (t) => {
   const responseBody = 'hello dave'
   const server = helper.startServer({ body: responseBody })
 
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     expectBody: responseBody,
     maxOverallRequests: 10
@@ -319,7 +319,7 @@ test('run should accept a unix socket/windows pipe', (t) => {
 
   helper.startServer({ socketPath })
 
-  run({
+  initJob({
     url: 'localhost',
     socketPath,
     connections: 2,
@@ -352,7 +352,7 @@ for (let i = 1; i <= 5; i++) {
   test(`run should count all ${i}xx status codes`, (t) => {
     const server = helper.startServer({ statusCode: i * 100 + 2 })
 
-    run({
+    initJob({
       url: `http://localhost:${server.address().port}`,
       connections: 2,
       duration: 2
@@ -391,7 +391,7 @@ for (let i = 1; i <= 5; i++) {
 
 test('run should not modify default options', (t) => {
   const origin = Object.assign({}, defaultOptions)
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     duration: 2
@@ -405,7 +405,7 @@ test('run should not modify default options', (t) => {
 test('run will exclude non 2xx stats from latency and throughput averages if excludeErrorStats is true', (t) => {
   const server = helper.startServer({ statusCode: 404 })
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 2,
     duration: 2,
@@ -451,7 +451,7 @@ test('tracker will emit reqError with error message on timeout', (t) => {
 
   const server = helper.startTimeoutServer()
 
-  const tracker = run({
+  const tracker = initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 1,
     duration: 5,
@@ -472,7 +472,7 @@ test('tracker will emit reqError with error message on error', (t) => {
 
   const server = helper.startSocketDestroyingServer()
 
-  const tracker = run({
+  const tracker = initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 10,
     duration: 15,
@@ -496,7 +496,7 @@ test('tracker will emit reqMismatch when body does not match expectBody', (t) =>
 
   const expectBody = 'goodbye world'
 
-  const tracker = run({
+  const tracker = initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 10,
     duration: 15,
@@ -517,7 +517,7 @@ test('tracker will emit tick with current counter value', (t) => {
 
   const server = helper.startSocketDestroyingServer()
 
-  const tracker = run({
+  const tracker = initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 10,
     duration: 10
@@ -535,7 +535,7 @@ test('throw if connections is greater than amount', (t) => {
   const server = helper.startSocketDestroyingServer()
 
   t.throws(function () {
-    run({
+    initJob({
       url: `http://localhost:${server.address().port}`,
       connections: 10,
       amount: 1,
@@ -545,7 +545,7 @@ test('throw if connections is greater than amount', (t) => {
 })
 
 test('run promise', (t) => {
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     duration: 2,
@@ -610,7 +610,7 @@ test('run promise', (t) => {
 test('should throw if duration is not a number nor a string', t => {
   t.plan(1)
   const server = helper.startServer()
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     duration: ['foobar'],
@@ -627,7 +627,7 @@ test('should throw if duration is not a number nor a string', t => {
 test('should emit error', t => {
   t.plan(1)
   const server = helper.startServer()
-  const tracker = run({
+  const tracker = initJob({
     url: `http://unknownhost:${server.address().port}`,
     connections: 1,
     timeout: 100,
@@ -649,7 +649,7 @@ test('should emit error', t => {
 test('should throw if timeout is less than zero', t => {
   t.plan(1)
   const server = helper.startServer()
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     timeout: -1,
@@ -666,7 +666,7 @@ test('should throw if timeout is less than zero', t => {
 test('should handle duration in string format', t => {
   t.plan(1)
   const server = helper.startServer()
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 2,
     duration: '1',
@@ -679,7 +679,7 @@ test('should handle duration in string format', t => {
 test('should count resets', t => {
   t.plan(1)
   const server = helper.startServer()
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 1,
     amount: 10,
@@ -702,7 +702,7 @@ test('should get onResponse callback invoked even when there is no body', async
   t.plan(4)
   const server = helper.startServer({ responses: [{ statusCode: 200, body: 'ok' }, { statusCode: 204 }] })
 
-  await run({
+  await initJob({
     url: 'http://localhost:' + server.address().port,
     connections: 1,
     amount: 2,
@@ -733,7 +733,7 @@ test('should use request from HAR', (t) => {
   const url = `http://localhost:${server.address().port}`
   const har = helper.customizeHAR('./fixtures/httpbin-get.json', 'https://httpbin.org', url)
 
-  run({
+  initJob({
     url,
     duration: 1,
     har
@@ -754,7 +754,7 @@ test('should use extend headers of HAR requests', (t) => {
   const url = `http://localhost:${server.address().port}`
   const har = helper.customizeHAR('./fixtures/httpbin-simple-get.json', 'https://httpbin.org', url)
 
-  run({
+  initJob({
     url,
     connections: 1,
     amount: 2,
@@ -779,7 +779,7 @@ test('should not override method or body of HAR requests', (t) => {
   const url = `http://localhost:${server.address().port}`
   const har = helper.customizeHAR('./fixtures/httpbin-simple-get.json', 'https://httpbin.org', url)
 
-  run({
+  initJob({
     url,
     connections: 1,
     amount: 2,
@@ -806,7 +806,7 @@ test('should ignore HAR requests targetting a different domain', (t) => {
   const url = `http://localhost:${server.address().port}`
   const har = helper.customizeHAR('./fixtures/multi-domains.json', 'https://httpbin.org', url)
 
-  run({
+  initJob({
     url,
     connections: 1,
     amount: 4,
@@ -826,7 +826,7 @@ test('should ignore HAR requests targetting a different domain', (t) => {
 test('should throw on invalid HAR', (t) => {
   t.plan(1)
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 1,
     amount: 4,
@@ -855,46 +855,3 @@ test('should throw on invalid HAR', (t) => {
     t.end()
   })
 })
-
-test('with workers', t => {
-  t.plan(3)
-
-  t.test('should error when using setupClient', t => {
-    t.plan(1)
-    const server = helper.startServer()
-    run({
-      url: 'http://localhost:' + server.address().port,
-      workers: 2,
-      setupClient: c => c
-    }).catch((err) => {
-      t.equal(err.message, 'Can not use setupClient option while using workers')
-      t.end()
-    })
-  })
-
-  t.test('should error when using request[x].setupRequest', t => {
-    t.plan(1)
-    const server = helper.startServer()
-    run({
-      url: 'http://localhost:' + server.address().port,
-      workers: 2,
-      requests: [{ setupRequest: () => {} }]
-    }).catch((err) => {
-      t.equal(err.message, 'Can not use setupRequest/onResponse options while using workers')
-      t.end()
-    })
-  })
-
-  t.test('should error when using request[x].onResponse', t => {
-    t.plan(1)
-    const server = helper.startServer()
-    run({
-      url: 'http://localhost:' + server.address().port,
-      workers: 2,
-      requests: [{ onResponse: () => {} }]
-    }).catch((err) => {
-      t.equal(err.message, 'Can not use setupRequest/onResponse options while using workers')
-      t.end()
-    })
-  })
-})
diff --git a/test/runAmount.test.js b/test/runAmount.test.js
index dcd562f..8a0e87a 100644
--- a/test/runAmount.test.js
+++ b/test/runAmount.test.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const test = require('tap').test
-const run = require('../lib/run')
+const initJob = require('../lib/init')
 const helper = require('./helper')
 const timeoutServer = helper.startTimeoutServer()
 const server = helper.startServer()
@@ -11,7 +11,7 @@ test('run should only send the expected number of requests', (t) => {
 
   let done = false
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     duration: 1,
     connections: 100,
@@ -27,7 +27,7 @@ test('run should only send the expected number of requests', (t) => {
     t.notOk(done)
   }, 1000)
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 2,
     maxConnectionRequests: 10
@@ -37,7 +37,7 @@ test('run should only send the expected number of requests', (t) => {
     t.equal(res.requests.sent, 20, 'totalRequests should match the expected amount')
   })
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 2,
     maxOverallRequests: 10
@@ -51,7 +51,7 @@ test('run should only send the expected number of requests', (t) => {
 test('should shutdown after all amounts timeout', (t) => {
   t.plan(5)
 
-  run({
+  initJob({
     url: `http://localhost:${timeoutServer.address().port}`,
     amount: 10,
     timeout: 2,
@@ -69,7 +69,7 @@ test('should reconnect twice to the server with a reset rate of 10 for 20 connec
   t.plan(3)
   const testServer = helper.startServer()
 
-  run({
+  initJob({
     url: 'localhost:' + testServer.address().port,
     connections: 1,
     amount: 20,
diff --git a/test/runMultiServer.test.js b/test/runMultiServer.test.js
index d76a47a..f288aef 100644
--- a/test/runMultiServer.test.js
+++ b/test/runMultiServer.test.js
@@ -1,5 +1,5 @@
 const test = require('tap').test
-const run = require('../lib/run')
+const initJob = require('../lib/init')
 const helper = require('./helper')
 
 const server1 = helper.startServer({ body: 'from server1' })
@@ -9,7 +9,7 @@ const server3 = helper.startServer({ body: 'from server3' })
 test('should receive the message from different server', (t) => {
   t.plan(3)
 
-  const instance = run({
+  const instance = initJob({
     url: [
       server1,
       server2,
diff --git a/test/runMultipart.test.js b/test/runMultipart.test.js
index 5ee5fe2..e521f70 100644
--- a/test/runMultipart.test.js
+++ b/test/runMultipart.test.js
@@ -5,9 +5,10 @@ const { tmpdir } = require('os')
 const { join } = require('path')
 const { writeFile } = require('fs')
 const { promisify } = require('util')
-const run = require('../lib/run')
+const initJob = require('../lib/init')
 const helper = require('./helper')
 const writef = promisify(writeFile)
+const hasWorkerSupport = require('./utils/has-worker-support')
 
 test('run should return an error with invalid form options', async t => {
   const cases = [
@@ -59,7 +60,7 @@ test('run should return an error with invalid form options', async t => {
   for (const c of cases) {
     t.test(c.name, async t => {
       const [err] = await new Promise((resolve) => {
-        run({
+        initJob({
           url: 'http://localhost:' + server.address().port,
           connections: 1,
           amount: 1,
@@ -108,20 +109,22 @@ test('run should take form options as a JSON string or a JS Object', async t =>
       value: form
     }
   ]
+  const allCases = [...cases, ...cases.map(c => ({ ...c, workers: true }))]
 
-  for (const c of cases) {
-    t.test(c.name, async t => {
+  for (const c of allCases) {
+    t.test(c.name, { skip: c.workers && !hasWorkerSupport }, async t => {
       const server = helper.startMultipartServer(null, payload => {
         t.equal('j5', payload.name)
         t.equal('j5.jpeg', payload.image.filename)
       })
       t.tearDown(() => server.close())
       const [err, res] = await new Promise((resolve) => {
-        run({
+        initJob({
           url: 'http://localhost:' + server.address().port,
           connections: 1,
           amount: 1,
-          form: c.value
+          form: c.value,
+          workers: c.workers ? 1 : undefined // use only one worker coz we're checking for 1 req
         }, (err, res) => {
           resolve([err, res])
         })
@@ -154,7 +157,7 @@ test('run should use a custom method if `options.method` is passed', t => {
       value: 'j5'
     }
   }
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     method: 'PUT',
     connections: 1,
@@ -186,7 +189,7 @@ test('run should set filename', t => {
       value: 'j5'
     }
   }
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     method: 'POST',
     connections: 1,
@@ -221,7 +224,7 @@ test('run should allow overriding filename', t => {
       value: 'j5'
     }
   }
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     method: 'POST',
     connections: 1,
@@ -256,7 +259,7 @@ test('run should allow overriding filename with file path', t => {
       value: 'j5'
     }
   }
-  run({
+  initJob({
     url: 'http://localhost:' + server.address().port,
     method: 'POST',
     connections: 1,
diff --git a/test/runRate.test.js b/test/runRate.test.js
index 6ae7335..77210a6 100644
--- a/test/runRate.test.js
+++ b/test/runRate.test.js
@@ -1,14 +1,14 @@
 'use strict'
 
 const test = require('tap').test
-const run = require('../lib/run')
+const initJob = require('../lib/init')
 const helper = require('./helper')
 const server = helper.startServer()
 
 test('run should only send the expected number of requests per second', (t) => {
   t.plan(9)
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 2,
     overallRate: 10,
@@ -19,7 +19,7 @@ test('run should only send the expected number of requests per second', (t) => {
     t.equal(res.requests.average, 10, 'should have sent 10 requests per second on average')
   })
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 2,
     connectionRate: 10,
@@ -30,7 +30,7 @@ test('run should only send the expected number of requests per second', (t) => {
     t.equal(res.requests.average, 20, 'should have sent 20 requests per second on average with two connections')
   })
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 15,
     overallRate: 10,
@@ -45,7 +45,7 @@ test('run should only send the expected number of requests per second', (t) => {
 test('run should compensate for coordinated omission when the expected number of requests per second is too high', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 100,
     connectionRate: 1000,
@@ -59,7 +59,7 @@ test('run should compensate for coordinated omission when the expected number of
 test('run should not compensate for coordinated omission when this feature is disabled', (t) => {
   t.plan(2)
 
-  run({
+  initJob({
     url: `http://localhost:${server.address().port}`,
     connections: 100,
     connectionRate: 1000,
diff --git a/test/serial/run.test.js b/test/serial/run.test.js
index 8690658..79e34a9 100644
--- a/test/serial/run.test.js
+++ b/test/serial/run.test.js
@@ -1,6 +1,6 @@
 'use strict'
 const test = require('tap').test
-const run = require('../../lib/run')
+const initJob = require('../../lib/init')
 
 test('should log error on connection error', t => {
   t.plan(1)
@@ -8,7 +8,7 @@ test('should log error on connection error', t => {
     t.type(obj, Error)
     console.error = () => {}
   }
-  run({
+  initJob({
     url: 'http://unknownhost',
     connections: 2,
     duration: 5,
diff --git a/test/serial/wasm.test.js b/test/serial/wasm.test.js
index 28c9ff6..74162cf 100644
--- a/test/serial/wasm.test.js
+++ b/test/serial/wasm.test.js
@@ -1,13 +1,13 @@
 'use strict'
 
 const test = require('tap').test
-const run = require('../../lib/run')
+const initJob = require('../../lib/init')
 const helper = require('../helper')
 
 test('should clean up HdrHistogram WASM memory at each run', async (t) => {
   const server = helper.startServer()
   const runTwentyTimes = (resolve, reject, numberOfRuns = 0) => {
-    run({
+    initJob({
       url: 'http://localhost:' + server.address().port,
       connections: 1,
       amount: 1
diff --git a/test/utils/has-worker-support.js b/test/utils/has-worker-support.js
new file mode 100644
index 0000000..298371b
--- /dev/null
+++ b/test/utils/has-worker-support.js
@@ -0,0 +1,5 @@
+'use strict'
+
+const semver = require('semver')
+
+module.exports = semver.gte(process.versions.node, '11.7.0')
diff --git a/test/utils/on-response.js b/test/utils/on-response.js
new file mode 100644
index 0000000..d0f5c2a
--- /dev/null
+++ b/test/utils/on-response.js
@@ -0,0 +1,5 @@
+'use strict'
+
+module.exports = (status, body, context) => {
+  context.foo = 'bar=baz'
+}
diff --git a/test/utils/setup-client.js b/test/utils/setup-client.js
new file mode 100644
index 0000000..0551578
--- /dev/null
+++ b/test/utils/setup-client.js
@@ -0,0 +1,5 @@
+'use strict'
+
+module.exports = (client) => {
+  client.setHeaders({ custom: 'my-header' })
+}
diff --git a/test/utils/setup-request.js b/test/utils/setup-request.js
new file mode 100644
index 0000000..01065f0
--- /dev/null
+++ b/test/utils/setup-request.js
@@ -0,0 +1,6 @@
+'use strict'
+
+module.exports = (req, context) => ({
+  ...req,
+  path: `/test-123?some=thing&${context.foo}`
+})
diff --git a/test/workers.test.js b/test/workers.test.js
new file mode 100644
index 0000000..8dd960d
--- /dev/null
+++ b/test/workers.test.js
@@ -0,0 +1,74 @@
+'use strict'
+
+const path = require('path')
+const test = require('tap').test
+const http = require('http')
+const initJob = require('../lib/init')
+const hasWorkerSupport = require('./utils/has-worker-support')
+
+test('setupRequest and onResponse work with workers', { skip: !hasWorkerSupport }, (t) => {
+  const server = http.createServer((req, res) => {
+    // it's not easy to assert things within setupRequest and onResponse
+    // when in workers mode. So, we set something in onResponse and use in the
+    // next Request and make sure it exist or we return 404.
+    if (req.method === 'GET' && req.url !== '/test-123?some=thing&bar=baz') {
+      res.statusCode = 404
+      res.end('NOT OK')
+      return
+    }
+
+    res.end('OK')
+  })
+  server.listen(0)
+  server.unref()
+
+  initJob({
+    url: 'http://localhost:' + server.address().port,
+    connections: 2,
+    amount: 4,
+    workers: 1,
+    requests: [
+      {
+        method: 'PUT',
+        onResponse: path.join(__dirname, './utils/on-response')
+      },
+      {
+        method: 'GET',
+        setupRequest: path.join(__dirname, './utils/setup-request')
+      }
+    ]
+  }, function (err, result) {
+    t.error(err)
+
+    t.equal(4, result['2xx'], 'should have 4 ok requests')
+    t.equal(0, result['4xx'], 'should not have any 404s')
+    t.end()
+  })
+})
+
+test('setupClient works with workers', { skip: !hasWorkerSupport }, (t) => {
+  const server = http.createServer((req, res) => {
+    if (req.headers.custom !== 'my-header') {
+      res.statusCode = 404
+      res.end('NOT OK')
+      return
+    }
+    res.end('OK')
+  })
+  server.listen(0)
+  server.unref()
+
+  initJob({
+    url: 'http://localhost:' + server.address().port,
+    connections: 2,
+    amount: 2,
+    workers: 1,
+    setupClient: path.join(__dirname, './utils/setup-client')
+  }, function (err, result) {
+    t.error(err)
+
+    t.equal(2, result['2xx'], 'should have 2 ok requests')
+    t.equal(0, result['4xx'], 'should not have any 404s')
+    t.end()
+  })
+})