amqp-connection-manager - npm Package Compare versions

Comparing version 1.1.1 to 1.1.2

lib/ChannelWrapper.js

@@ -32,2 +32,4 @@ // Generated by CoffeeScript 1.10.0
       this._working = false;
+      this._settingUp = null;
+      this._channel = null;
       this._workerNumber = 0;
@@ -53,17 +55,26 @@ this._setups = [];
         return function(channel) {
-          _this._channelInProgress = channel;
-          return Promise.all(_this._setups.map(function(setupFn) {
+          _this._channel = channel;
+          channel.on('close', function() {
+            return _this._onChannelClose(channel);
+          });
+          return _this._settingUp = Promise.all(_this._setups.map(function(setupFn) {
             return pb.callFn(setupFn, 1, null, channel)["catch"](function(err) {
-              return _this.emit('error', err, {
-                name: _this.name
-              });
+              if (_this._channel) {
+                return _this.emit('error', err, {
+                  name: _this.name
+                });
+              } else {
+
+              }
             });
           })).then(function() {
-            return channel;
+            _this._settingUp = null;
+            return _this._channel;
           });
         };
       })(this)).then((function(_this) {
-        return function(channel) {
-          _this._channelInProgress = null;
-          _this._channel = channel;
+        return function() {
+          if (_this._channel == null) {
+            return;
+          }
           _this._startWorker();
@@ -74,5 +85,7 @@ return _this.emit('connect');
         return function(err) {
-          return _this.emit('error', err, {
+          _this.emit('error', err, {
             name: _this.name
           });
+          _this._settingUp = null;
+          return _this._channel = null;
         };
@@ -82,4 +95,11 @@ })(this));
 
+    ChannelWrapper.prototype._onChannelClose = function(channel) {
+      if (this._channel === channel) {
+        return this._channel = null;
+      }
+    };
+
     ChannelWrapper.prototype._onDisconnect = function() {
       this._channel = null;
+      this._settingUp = null;
       return this._working = false;
@@ -93,3 +113,5 @@ };
           if (_this._channel) {
-            return setup(_this._channel);
+            return (_this._settingUp || Promise.resolve()).then(function() {
+              return setup(_this._channel);
+            });
           }
@@ -102,4 +124,8 @@ };
       this._setups = _.without(this._setups, setup);
-      if (this._channel && (teardown != null)) {
-        return pb.callFn(teardown, 1, null, this._channel);
+      if (this._channel) {
+        return (this._settingUp || Promise.resolve()).then((function(_this) {
+          return function() {
+            return pb.callFn(teardown, 1, null, _this._channel);
+          };
+        })(this));
       }
@@ -114,5 +140,6 @@ });
       var ref1;
+      this._working = false;
       if (this._messages.length !== 0) {
         this._messages.forEach(function(message) {
-          return message.reject();
+          return message.reject(new Error('Channel closed'));
         });
@@ -130,3 +157,3 @@ }
     ChannelWrapper.prototype.waitForConnect = pb["break"](function() {
-      if (this._channel) {
+      if (this._channel && !this._settingUp) {
         return Promise.resolve();
@@ -142,4 +169,8 @@ } else {
 
+    ChannelWrapper.prototype._shouldPublish = function() {
+      return (this._messages.length > 0) && !this._settingUp && this._channel;
+    };
+
     ChannelWrapper.prototype._startWorker = function() {
-      if (this._channel && !this._working) {
+      if (!this._working && this._shouldPublish()) {
         this._working = true;
@@ -153,3 +184,3 @@ this._workerNumber++;
       var channel, message;
-      if ((this._messages.length === 0) || !this._channel || !this._working || (workerNumber !== this._workerNumber)) {
+      if (!this._shouldPublish() || !this._working || (workerNumber !== this._workerNumber)) {
         this._working = false;
@@ -201,13 +232,15 @@ return Promise.resolve();
           _this._messages.shift();
-          return message.resolve(result);
+          message.resolve(result);
+          return _this._publishQueuedMessages(workerNumber);
         };
       })(this), (function(_this) {
         return function(err) {
-          _this._messages.shift();
-          return message.reject(err);
+          if (!_this._channel) {
+
+          } else {
+            _this._messages.shift();
+            message.reject(err);
+            return _this._publishQueuedMessages(workerNumber);
+          }
         };
-      })(this)).then((function(_this) {
-        return function() {
-          return _this._publishQueuedMessages(workerNumber);
-        };
       })(this))["catch"]((function(_this) {
@@ -226,17 +259,11 @@ return function(err) {
     ChannelWrapper.prototype.ack = function() {
-      var args, channel, ref1;
+      var args, ref1;
       args = 1 <= arguments.length ? slice.call(arguments, 0) : [];
-      channel = (ref1 = this._channelInProgress) != null ? ref1 : this._channel;
-      if (channel) {
-        return channel.ack.apply(channel, args);
-      }
+      return (ref1 = this._channel) != null ? ref1.ack.apply(ref1, args) : void 0;
     };
 
     ChannelWrapper.prototype.nack = function() {
-      var args, channel, ref1;
+      var args, ref1;
       args = 1 <= arguments.length ? slice.call(arguments, 0) : [];
-      channel = (ref1 = this._channelInProgress) != null ? ref1 : this._channel;
-      if (channel) {
-        return channel.nack.apply(channel, args);
-      }
+      return (ref1 = this._channel) != null ? ref1.nack.apply(ref1, args) : void 0;
     };
@@ -243,0 +270,0 @@

package.json

 {
   "name": "amqp-connection-manager",
-  "version": "1.1.1",
+  "version": "1.1.2",
   "description": "Auto-reconnect and round robin support for amqplib.",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

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

24499

increased by3.53%

Lines of code

391

increased by6.25%

No dependency changes