nats - npm Package Compare versions

Comparing version 0.7.29 to 0.8.0

lib/nats.js

@@ -24,3 +24,3 @@ /*!
  */
-var VERSION = '0.7.29',
+var VERSION = '0.8.0',
 
@@ -152,3 +152,3 @@ DEFAULT_PORT = 4222,
  *
- * @param {Mixed} opts
+ * @param {Mixed} [opts]
  * @api public
@@ -168,3 +168,3 @@ */
  *
- * @params {Mixed} opts
+ * @params {Mixed} [opts]
  *
@@ -211,3 +211,3 @@ * @api public
  *
- * @param {Mixed} opts
+ * @param {Mixed} [opts]
  * @api private
@@ -217,12 +217,13 @@ */
     var options = this.options = {
-        'verbose': false,
-        'pedantic': false,
-        'reconnect': true,
-        'maxReconnectAttempts': DEFAULT_MAX_RECONNECT_ATTEMPTS,
-        'reconnectTimeWait': DEFAULT_RECONNECT_TIME_WAIT,
-        'encoding': 'utf8',
-        'tls': false,
-        'waitOnFirstConnect': false,
-        'pingInterval': DEFAULT_PING_INTERVAL,
-        'maxPingOut': DEFAULT_MAX_PING_OUT,
+        verbose: false,
+        pedantic: false,
+        reconnect: true,
+        maxReconnectAttempts: DEFAULT_MAX_RECONNECT_ATTEMPTS,
+        reconnectTimeWait: DEFAULT_RECONNECT_TIME_WAIT,
+        encoding: 'utf8',
+        tls: false,
+        waitOnFirstConnect: false,
+        pingInterval: DEFAULT_PING_INTERVAL,
+        maxPingOut: DEFAULT_MAX_PING_OUT,
+        useOldRequestStyle: false
     };
@@ -268,2 +269,3 @@
         this.assignOption(opts, 'maxPingOut');
+        this.assignOption(opts, 'useOldRequestStyle');
     }
@@ -523,3 +525,3 @@
         'pedantic': this.options.pedantic,
-        'protocol': 1,
+        'protocol': 1
     };
@@ -1087,3 +1089,3 @@ if (this.user !== undefined) {
  *
- * @param {Function} opt_callback
+ * @param {Function} [opt_callback]
  * @api public
@@ -1111,5 +1113,5 @@ */
  * @param {String} subject
- * @param {String} opt_msg
- * @param {String} opt_reply
- * @param {Function} opt_callback
+ * @param {String} [msg]
+ * @param {String} [opt_reply]
+ * @param {Function} [opt_callback]
  * @api public
@@ -1193,3 +1195,3 @@ */
  * @param {String} subject
- * @param {Object} opts
+ * @param {Object} [opts]
  * @param {Function} callback
@@ -1242,3 +1244,3 @@ * @return {Mixed}
  * @param {Mixed} sid
- * @param {Number} opt_max
+ * @param {Number} [opt_max]
  * @api public
@@ -1251,2 +1253,10 @@ */
 
+    // in the case of new muxRequest, it is possible they want perform
+    // an unsubscribe with the returned 'sid'. Intercept that and clear
+    // the request configuration. Mux requests are always negative numbers
+    if(sid < 0) {
+        this.cancelMuxRequest(sid);
+        return;
+    }
+
     var proto;
@@ -1283,2 +1293,3 @@ if (opt_max) {
  * @param {Number} expected
+ * @param {Function} callback
  * @api public
@@ -1308,12 +1319,22 @@ */
  * will need to unsubscribe to stop the message stream.
+ *
+ * You can also optionally specify the number of milliseconds to wait for the messages
+ * to receive using opt_options = {timeout: N}. When the number of messages specified
+ * is received before a timeout, the subscription auto-cancels. If the number of messages
+ * is not specified, it is the responsibility of the client to unsubscribe to prevent
+ * a timeout.
+ *
  * The Subscriber Id is returned.
  *
  * @param {String} subject
- * @param {String} opt_msg
- * @param {Object} opt_options
+ * @param {String} [opt_msg]
+ * @param {Object} [opt_options]
  * @param {Function} callback
- * @return {Mixed}
+ * @return {Number}
  * @api public
  */
 Client.prototype.request = function(subject, opt_msg, opt_options, callback) {
+    if(this.options.useOldRequestStyle) {
+        return this.oldRequest(subject, opt_msg, opt_options, callback);
+    }
     if (typeof opt_msg === 'function') {
@@ -1328,2 +1349,35 @@ callback = opt_msg;
     }
+
+    opt_options = opt_options || {};
+    var conf = this.initMuxRequestDetails(callback, opt_options.max);
+    this.publish(subject, opt_msg, conf.inbox);
+
+    if(opt_options.timeout) {
+        var client = this;
+        conf.timer = setTimeout(function() {
+            if(conf.callback) {
+                conf.callback(new NatsError(REQ_TIMEOUT_MSG_PREFIX + conf.id, REQ_TIMEOUT));
+            }
+            client.cancelMuxRequest(conf.token);
+        }, opt_options.timeout);
+    }
+
+    return conf.id;
+};
+
+
+/**
+ * @deprecated
+ * @api private
+ */
+Client.prototype.oldRequest = function(subject, opt_msg, opt_options, callback) {
+    if (typeof opt_msg === 'function') {
+        callback = opt_msg;
+        opt_msg = EMPTY;
+        opt_options = null;
+    }
+    if (typeof opt_options === 'function') {
+        callback = opt_options;
+        opt_options = null;
+    }
     var inbox = createInbox();
@@ -1349,11 +1403,156 @@ var s = this.subscribe(inbox, opt_options, function(msg, reply) {
  * @param {String} subject
- * @param {String} opt_msg
- * @param {Object} opt_options
+ * @param {String} [opt_msg]
+ * @param {Object} [opt_options]
+ * @param {Number} timeout
  * @param {Function} callback - can be called with message or NatsError if the request timed out.
- * @return {Mixed}
+ * @return {Number}
  * @api public
  */
 Client.prototype.requestOne = function(subject, opt_msg, opt_options, timeout, callback) {
+    if(this.options.useOldRequestStyle) {
+        return this.oldRequestOne(subject, opt_msg, opt_options, timeout, callback);
+    }
+
+    if (typeof opt_msg === 'number') {
+        timeout = opt_msg;
+        callback = opt_options;
+        opt_msg = EMPTY;
+        opt_options = null;
+    }
+
+    if (typeof opt_options === 'number') {
+        timeout = opt_options;
+        callback = timeout;
+        opt_options = null;
+    }
+
     opt_options = opt_options || {};
     opt_options.max = 1;
+    opt_options.timeout = timeout;
+
+    return this.request(subject, opt_msg, opt_options, callback);
+};
+
+/**
+ * Strips the prefix of the request reply to derive the token.
+ * This is internal and only used by the new requestOne.
+ *
+ * @api private
+ */
+Client.prototype.extractToken = function(subject) {
+    return subject.substr(this.respmux.inboxPrefixLen);
+};
+
+/**
+ * Creates a subscription for the global inbox in the new requestOne.
+ * Request tokens, timer, and callbacks are tracked here.
+ *
+ * @api private
+ */
+Client.prototype.createResponseMux = function() {
+    if(!this.respmux) {
+        var client = this;
+        var inbox = createInbox();
+        var ginbox = inbox + ".*";
+        var sid = this.subscribe(ginbox, function(msg, reply, subject) {
+            var token = client.extractToken(subject);
+            var conf = client.getMuxRequestConfig(token);
+            if(conf) {
+                if(conf.hasOwnProperty('max_messages')) {
+                    conf.max_messages--;
+                    if (conf.max_messages <= 0) {
+                        client.cancelMuxRequest(token);
+                    }
+                }
+                conf.callback(msg);
+            }
+        });
+        this.respmux = {};
+        this.respmux.inbox = inbox;
+        this.respmux.inboxPrefixLen = inbox.length + 1;
+        this.respmux.subscriptionID = sid;
+        this.respmux.requestMap = {};
+        this.respmux.nextID = -1;
+    }
+    return this.respmux.inbox;
+};
+
+/**
+ * Stores the request callback and other details
+ *
+ * @api private
+ */
+Client.prototype.initMuxRequestDetails = function(callback, max_messages) {
+    var ginbox = this.createResponseMux();
+    var token = nuid.next();
+    var inbox = ginbox + '.' + token;
+
+    var conf = {token: token, callback: callback, inbox: inbox, id: this.respmux.nextID--};
+    if(max_messages > 0) {
+        conf.max_messages = max_messages;
+    }
+
+    this.respmux.requestMap[token] = conf;
+    return conf;
+};
+
+Client.prototype.getMuxRequestConfig = function(token) {
+    return this.respmux.requestMap[token];
+};
+
+/**
+ * Cancels the mux request
+ *
+ * @api private
+ */
+Client.prototype.cancelMuxRequest = function(token) {
+
+    // if the token is a number, we have a fake sid, find the request
+    if (typeof token === 'number') {
+        var entry = null;
+        for (var p in this.respmux.requestMap) {
+            if (this.respmux.requestMap.hasOwnProperty(p)) {
+                var v = this.respmux.requestMap[p];
+                if (v.id === token) {
+                    entry = v;
+                    break;
+                }
+            }
+        }
+        if (entry) {
+            token = entry.token;
+        }
+    }
+
+    var conf = this.respmux.requestMap[token];
+    if (conf) {
+        if(conf.timer) {
+            clearTimeout(conf.timer);
+        }
+        delete this.respmux.requestMap[token];
+    }
+    return conf;
+};
+
+/**
+ * @deprecated
+ * @api private
+ */
+Client.prototype.oldRequestOne = function(subject, opt_msg, opt_options, timeout, callback) {
+    if (typeof opt_msg === 'number') {
+        timeout = opt_msg;
+        callback = opt_options;
+        opt_msg = EMPTY;
+        opt_options = null;
+    }
+
+    if (typeof opt_options === 'number') {
+        timeout = opt_options;
+        callback = timeout;
+        opt_options = null;
+    }
+
+    opt_options = opt_options || {};
+    opt_options.max = 1;
+
     var sid = this.request(subject, opt_msg, opt_options, callback);
@@ -1360,0 +1559,0 @@ this.timeout(sid, timeout, 1, function() {

package.json

 {
   "name": "nats",
-  "version": "0.7.29",
+  "version": "0.8.0",
   "description": "Node.js client for NATS, a lightweight, high-performance cloud native messaging system",
@@ -5,0 +5,0 @@ "keywords": [

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

116670

increased by96.43%

Number of package files

8

increased by14.29%

Lines of code

3037

increased by115.7%

Worsened metrics

Number of medium supply chain risk alerts

4

increased by100%

No dependency changes