feat(sender-link): add support for sending with no reply

This allows users to send messages on a SenderLink (and the
corresponding SenderLinkStream) without requiring a Promise to be
returned. There are no guarantees regarding unsettled sends or
tracking for pending sends with this options set, the goal here is
purely for performance.

lib/amqp_client.js

@@ -256,7 +256,7 @@ AMQPClient.prototype.createSender = function(address, policyOverrides) {
  */
 AMQPClient.prototype.createSenderStream = function(address, policyOverrides) {
   return this.createSender(address, policyOverrides)
-    .then(function(link) { return new SenderStream(link); });
+    .then(function(link) { return new SenderStream(link, policyOverrides); });
 };
 
 /**

lib/policies/policy_utilities.js

@@ -57,14 +57,22 @@ var CreditPolicies = {
 
 module.exports.CreditPolicies = CreditPolicies;
 
-var SenderCallbackPolicies = {
-  // Only callback when settled Disposition received from recipient
+/**
+ * Defines the behavior of the return value of `SenderLink.send`
+ * @enum
+ */
+var SenderCallbackPolicy = {
+  /** Callback immediately after sending, no promise is created */
+  None: 'none',
+  /** Only callback when settled Disposition received from recipient */
   OnSettle: 'settled',
-  // Callback as soon as sent, will not call-back again if future disposition results in error.
-  OnSent: 'sent'
+  /** Callback as soon as sent, will not call-back again if future disposition
+   *  results in error. */
+  OnSent: 'sent',
 };
 
-module.exports.SenderCallbackPolicies = SenderCallbackPolicies;
+module.exports.SenderCallbackPolicies = SenderCallbackPolicy; // deprecated
+module.exports.SenderCallbackPolicy = SenderCallbackPolicy;
 
 function merge(newPolicy, base) {
   return u.deepMerge(newPolicy, base);

lib/sender_link.js

@@ -53,13 +53,17 @@ SenderLink.prototype.canSend = function() {
  * @inner @memberof SenderLink
  * @param {object|string|array} msg Message to send.  Will be encoded using sender link policy's encoder.
  * @param {object} [options] An object of options to attach to the message including: annotations, properties,
-                             and application properties
+ *                           and application properties
+ * @param {string} [options.callback] Determines when the send operation should callback. Possible
+ *                                    options are: 'sent', 'settled' and 'none'. For the best performance
+ *                                    choose 'none', which is essentially "send and forget" and notably will
+ *                                    not return a promise.
  * @param {object} [options.annotations] Annotations for the message, if any.  See AMQP spec for details, and server for specific
  *                                       annotations that might be relevant (e.g. x-opt-partition-key on EventHub).  If node-amqp-encoder'd
  *                                       map is given, it will be translated to appropriate internal types.  Simple maps will be converted
  *                                       to AMQP Fields type as defined in the spec.
  *
- * @return {Promise}
+ * @return {Promise|null}
  */
 SenderLink.prototype.send = function(msg, options) {
   if (!this.session.connection || this.session.connection.connected === false) {
@@ -76,24 +80,44 @@ SenderLink.prototype.send = function(msg, options) {
       options.properties.subject = this.policy.defaultSubject;
   }
 
-  var self = this;
-  return new Promise(function(resolve, reject) {
-    var message;
-    if (u.isObject(msg) && msg.hasOwnProperty('body')) {
-      message = msg;
-      message.body = self.policy.encoder ? self.policy.encoder(message.body) : message.body;
-      u.merge(message, options);
+  var message;
+  var deliveryTag = this.session._deliveryTag++;
+  if (u.isObject(msg) && msg.hasOwnProperty('body')) {
+    message = msg;
+    message.body = this.policy.encoder ? this.policy.encoder(message.body) : message.body;
+    u.merge(message, options);
+  } else {
+    msg = this.policy.encoder ? this.policy.encoder(msg) : msg;
+    if (u.isObject(options)) {
+      options.body = msg;
+      message = options;
     } else {
-      msg = self.policy.encoder ? self.policy.encoder(msg) : msg;
-      if (u.isObject(options)) {
-        options.body = msg;
-        message = options;
-      } else {
-        message = { body: msg };
-      }
+      message = { body: msg };
     }
+  }
 
-    var deliveryTag = self.session._deliveryTag++;
+  var self = this,
+      cbPolicy = self.policy.callback;
+
+  if (cbPolicy === putils.SenderCallbackPolicies.None) {
+    var sendMessage = function(err) {
+      if (!!err) throw err;
+      debug('sending: ', msg);
+      return self._sendMessage(message, {
+        deliveryTag: new Buffer(deliveryTag.toString())
+      });
+    };
+
+    if (!this.canSend()) {
+      this._pendingSends.push(sendMessage);
+      return;
+    }
+
+    sendMessage();
+    return;
+  }
+
+  return new Promise(function(resolve, reject) {
     var sendMessage = function(err) {
       if (err) {
         return reject(err);
@@ -104,14 +128,10 @@ SenderLink.prototype.send = function(msg, options) {
         deliveryTag: new Buffer(deliveryTag.toString())
       });
 
-      var cbPolicy = self.policy.callback;
       if (cbPolicy === putils.SenderCallbackPolicies.OnSettle) {
         var deferredSender = function(err, state) {
-          if (!!err) {
-            reject(err);
-          } else {
-            resolve(state);
-          }
+          if (!!err) return reject(err);
+          resolve(state);
         };
         self._unsettledSends[messageId] = deferredSender;
       } else if (cbPolicy === putils.SenderCallbackPolicies.OnSent) {

lib/streams/sender_stream.js

@@ -1,15 +1,22 @@
 'use strict';
-var Writable = require('stream').Writable,
+var Promise = require('bluebird'),
+    Writable = require('stream').Writable,
     util = require('util');
 
 function SenderStream(link, options) {
   Writable.call(this, { objectMode: true });
+
+  options = options || {};
+  this._noReply = options.hasOwnProperty('noReply') ? !!options.noReply : false;
   this._link = link;
 }
 util.inherits(SenderStream, Writable);
 
 SenderStream.prototype._write = function(chunk, encoding, callback) {
-  return this._link.send(chunk)
+  var promise = this._link.send(chunk, { noReply: this._noReply });
+  if (!(promise instanceof Promise)) return callback();
+
+  return promise
     .then(function() { callback(); })
     .error(function(err) { callback(err); });
 };

test/integration/qpid/streams.test.js

@@ -29,13 +29,14 @@ describe('QPID', function() {
 
       Promise.all([
         test.client.createReceiverStream(config.defaultLink),
-        test.client.createSender(config.defaultLink)
+        test.client.createSender(config.defaultLink, { callback: 'none' })
       ])
       .spread(function(stream, sender) {
         var count = 0;
         stream.on('data', function(data) {
           expect(expected[count]).to.eql(data.body);
           count++;
+
           if (count === expected.length) done();
         });
 
@@ -61,7 +62,7 @@ describe('QPID', function() {
 
       Promise.all([
         test.client.createReceiver(config.defaultLink),
-        test.client.createSenderStream(config.defaultLink)
+        test.client.createSenderStream(config.defaultLink, { callback: 'none' })
       ])
       .spread(function(receiver, stream) {
         var count = 0;
@@ -111,7 +112,7 @@ describe('QPID', function() {
       Promise.all([
         test.client.createReceiver(config.defaultLink),
         test.client.createReceiverStream('test.streams.queue'),
-        test.client.createSenderStream(config.defaultLink),
+        test.client.createSenderStream(config.defaultLink, { callback: 'none' }),
       ])
       .spread(function(receiver, receiverStream, senderStream) {
         var count = 0;
@@ -122,7 +123,7 @@ describe('QPID', function() {
         });
 
         receiverStream.pipe(senderStream);
-        return test.client.createSender('test.streams.queue');
+        return test.client.createSender('test.streams.queue', { callback: 'none' });
       })
       .then(function(sender) {
         return Promise.mapSeries(expected, function(v) { return sender.send(v); });

test/integration/servicebus/eventhubs/streams.test.js

@@ -31,68 +31,68 @@ function teardown() {
 
 describe('ServiceBus', function() {
 
-  describe('Streams', function() {
-  describe('ReceiverStream', function() {
-    beforeEach(setup);
-    afterEach(teardown);
+describe('Streams', function() {
+describe('ReceiverStream', function() {
+  beforeEach(setup);
+  afterEach(teardown);
 
-    it('should let you create a receiver link as a readable stream', function(done) {
-      expect(config.partitionSenderLinkPrefix,
-        'Required env vars not found in ' + Object.keys(process.env)).to.exist;
+  it('should let you create a receiver link as a readable stream', function(done) {
+    expect(config.partitionSenderLinkPrefix,
+      'Required env vars not found in ' + Object.keys(process.env)).to.exist;
 
-      var dataString = u.uuidV4().replace(/-/g, ''),
-          expected = Array.apply(null, new Array(20))
-            .map(function(a) { return Math.floor(Math.random() * 100); });
+    var dataString = u.uuidV4().replace(/-/g, ''),
+        expected = Array.apply(null, new Array(20))
+          .map(function(a) { return Math.floor(Math.random() * 100); });
 
-      Promise.all([
-        test.client.createReceiverStream(config.receiverLinkPrefix + test.partition),
-        test.client.createSender(config.partitionSenderLinkPrefix + test.partition)
-      ])
-      .spread(function(stream, sender) {
-        var count = 0;
-        stream.on('data', function(data) {
-          if (data.body.DataString !== dataString) return;  // ignore previously run tests
-          expect(expected[count]).to.eql(data.body.DataValue);
-          count++;
-          if (count === expected.length) done();
-        });
+    Promise.all([
+      test.client.createReceiverStream(config.receiverLinkPrefix + test.partition),
+      test.client.createSender(config.partitionSenderLinkPrefix + test.partition, { callback: 'none' })
+    ])
+    .spread(function(stream, sender) {
+      var count = 0;
+      stream.on('data', function(data) {
+        if (data.body.DataString !== dataString) return;  // ignore previously run tests
+        expect(expected[count]).to.eql(data.body.DataValue);
+        count++;
+        if (count === expected.length) done();
+      });
 
-        return Promise.mapSeries(expected, function(v) {
-          return sender.send({ DataString: dataString, DataValue: v });
-        });
+      return Promise.mapSeries(expected, function(v) {
+        return sender.send({ DataString: dataString, DataValue: v });
       });
     });
-  }); // ReceiverStream
-
-  describe('SenderStream', function() {
-    beforeEach(setup);
-    afterEach(teardown);
+  });
+}); // ReceiverStream
 
-    it('should let you create a sender link as a writable stream', function(done) {
-      var dataString = u.uuidV4().replace(/-/g, ''),
-          expected = Array.apply(null, new Array(20))
-            .map(function(a) { return Math.floor(Math.random() * 100); });
+describe('SenderStream', function() {
+  beforeEach(setup);
+  afterEach(teardown);
 
-      Promise.all([
-        test.client.createReceiver(config.receiverLinkPrefix + test.partition),
-        test.client.createSenderStream(config.partitionSenderLinkPrefix + test.partition)
-      ])
-      .spread(function(receiver, stream) {
-        var count = 0;
-        receiver.on('message', function(data) {
-          if (data.body.DataString !== dataString) return;  // ignore previously run tests
-          expect(expected[count]).to.eql(data.body.DataValue);
-          count++;
-          if (count === expected.length) done();
-        });
+  it('should let you create a sender link as a writable stream', function(done) {
+    var dataString = u.uuidV4().replace(/-/g, ''),
+        expected = Array.apply(null, new Array(20))
+          .map(function(a) { return Math.floor(Math.random() * 100); });
 
-        for (var i = 0; i < expected.length; i++) {
-          stream.write({ DataString: dataString, DataValue: expected[i] });
-        }
+    Promise.all([
+      test.client.createReceiver(config.receiverLinkPrefix + test.partition),
+      test.client.createSenderStream(config.partitionSenderLinkPrefix + test.partition, { callback: 'none' })
+    ])
+    .spread(function(receiver, stream) {
+      var count = 0;
+      receiver.on('message', function(data) {
+        if (data.body.DataString !== dataString) return;  // ignore previously run tests
+        expect(expected[count]).to.eql(data.body.DataValue);
+        count++;
+        if (count === expected.length) done();
       });
+
+      for (var i = 0; i < expected.length; i++) {
+        stream.write({ DataString: dataString, DataValue: expected[i] });
+      }
     });
-  }); // SenderStream
+  });
+}); // SenderStream
 
-  }); // Streams
+}); // Streams
 
 }); // ServiceBus