fix: verify that options is an object

microfleet · Nov 13, 2019 · bef268a · bef268a
1 parent 8f72f0a
commit bef268a
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 30 deletions.
diff --git a/bench/roundtrip.js b/bench/roundtrip.js
@@ -1,7 +1,7 @@
 const Promise = require('bluebird');
 const Benchmark = require('benchmark');
-const AMQPTransport = require('../lib');
 const fmt = require('util').format;
+const AMQPTransport = require('../lib');
 
 const configuration = {
   exchange: 'test-exchange',
@@ -29,10 +29,11 @@ function listener(message, headers, actions, callback) {
 }
 
 // opts for consumer
-const opts = Object.assign({}, configuration, {
+const opts = {
+  ...configuration,
   queue: 'tq',
   listen: 'tq',
-});
+};
 
 // publisher
 const publisher = new AMQPTransport(configuration);
@@ -42,27 +43,27 @@ Promise.join(
   AMQPTransport.connect(opts, listener),
   publisher.connect()
 )
-.spread((consumer) => {
-  const suite = new Benchmark.Suite('RabbitMQ');
-  suite.add('Round-trip', {
-    defer: true,
-    fn: function test(deferred) {
-      return publisher
-        .publishAndWait('tq', 'tq')
-        .finally(() => {
-          messagesSent += 1;
-          deferred.resolve();
-        });
-    },
-  })
-  .on('complete', function suiteCompleted() {
-    const stats = this.filter('fastest')[0].stats;
-    const times = this.filter('fastest')[0].times;
-    process.stdout.write(fmt('Messages sent: %s\n', messagesSent));
-    process.stdout.write(fmt('Mean is %s ms ~ %s %\n', stats.mean * 1000, stats.rme));
-    process.stdout.write(fmt('Total time is %s s %s s\n', times.elapsed, times.period));
-    consumer.close();
-    publisher.close();
-  })
-  .run({ async: false, defer: true });
-});
+  .spread((consumer) => {
+    const suite = new Benchmark.Suite('RabbitMQ');
+    suite.add('Round-trip', {
+      defer: true,
+      fn: function test(deferred) {
+        return publisher
+          .publishAndWait('tq', 'tq')
+          .finally(() => {
+            messagesSent += 1;
+            deferred.resolve();
+          });
+      },
+    })
+      .on('complete', function suiteCompleted() {
+        const { stats } = this.filter('fastest')[0];
+        const { times } = this.filter('fastest')[0];
+        process.stdout.write(fmt('Messages sent: %s\n', messagesSent));
+        process.stdout.write(fmt('Mean is %s ms ~ %s %\n', stats.mean * 1000, stats.rme));
+        process.stdout.write(fmt('Total time is %s s %s s\n', times.elapsed, times.period));
+        consumer.close();
+        publisher.close();
+      })
+      .run({ async: false, defer: true });
+  });
diff --git a/src/amqp.js b/src/amqp.js
@@ -1037,7 +1037,9 @@ class AMQPTransport extends EventEmitter {
 
   _replyOptions(options = {}) {
     return {
-      simpleResponse: options.simpleResponse === undefined ? this._defaultOpts.simpleResponse : options.simpleResponse,
+      simpleResponse: options.simpleResponse === undefined
+        ? this._defaultOpts.simpleResponse
+        : options.simpleResponse,
     };
   }
 
@@ -1126,6 +1128,8 @@ class AMQPTransport extends EventEmitter {
    * @return {Promise}
    */
   async createMessageHandler(routing, message, options, publishMessage, span) {
+    assert(typeof options === 'object' && options !== null, 'options must be an object');
+
     const replyTo = options.replyTo || this._replyTo;
     const time = process.hrtime();
     const replyOptions = this._replyOptions(options);
@@ -1193,10 +1197,10 @@ class AMQPTransport extends EventEmitter {
         expiration: Math.ceil(timeout * 0.9).toString(),
       }, span)
       .tap(() => {
-        this.log.trace('message published in %s', latency(time));
+        this.log.trace({ latency: latency(time) }, 'message published');
       })
       .catch((err) => {
-        this.log.error('error sending message', err);
+        this.log.error({ err }, 'error sending message');
         replyStorage.reject(correlationId, err);
       });