Feature/bn add exchange options

lib/abstract_ascoltatore.js

@@ -128,7 +128,7 @@ AbstractAscoltatore.prototype._setPublish = function() {
     publish = this.publish;
   } else {
     publish = function (topic, payload, options, done) {
-      return f.call(this, topic, payload, done);
+      return f.call(this, topic, payload, options, done);
     };
   }
 

lib/amqp_ascoltatore.js

@@ -8,6 +8,12 @@ var AbstractAscoltatore = require('./abstract_ascoltatore');
 var steed = require("steed")();
 var SubsCounter = require("./subs_counter");
 var debug = require("debug")("ascoltatori:amqp");
+var _ = require('lodash');
+
+var defaultExchangeOptions = {
+  type: "topic",
+  confirm: true
+};
 
 /**
  * The AMQPAscoltatore is a class that inherits from AbstractAscoltatore.
@@ -80,10 +86,8 @@ AMQPAscoltatore.prototype._startConn = function() {
 
       function(callback) {
         debug("connected");
-        that._exchange = conn.exchange(that._opts.exchange, {
-          type: "topic",
-          confirm: true
-        });
+        var exchangeOptions = _.assignIn({}, defaultExchangeOptions, that._opts.exchangeOptions || {});
+        that._exchange = conn.exchange(that._opts.exchange, exchangeOptions);
         that._exchange.once("open", wrap(callback));
       },
 
@@ -149,12 +153,12 @@ AMQPAscoltatore.prototype.subscribe = function subscribe(topic, callback, done)
   debug("registered new subscriber for topic " + topic);
 };
 
-AMQPAscoltatore.prototype.publish = function publish(topic, message, done) {
+AMQPAscoltatore.prototype.publish = function publish(topic, message, options, done) {
   this._raiseIfClosed();
 
   debug("new message published to " + topic);
 
-  this._exchange.publish(this._pubTopic(topic), String(message));
+  this._exchange.publish(this._pubTopic(topic), String(message), options);
   defer(done);
 };
 

package.json

@@ -60,6 +60,7 @@
   },
   "dependencies": {
     "debug": "^2.2.0",
+    "lodash": "^4.0.0",
     "uuid": "^3.0.0",
     "qlobber": "~0.7.0",
     "steed": "^1.1.3"