Add code and unit test to handle: CONNECTION_FORCED - broker forced c…

…onnection closure with reason 'shutdown'
rudijs · May 27, 2015 · 766f77b · 766f77b
1 parent b22ff6a
commit 766f77b
Show file tree

Hide file tree

Showing 2 changed files with 76 additions and 3 deletions.
diff --git a/lib/rpc-publisher-factory.js b/lib/rpc-publisher-factory.js
@@ -75,6 +75,17 @@ var rpcPublisherProto = {
           this.publisherDomain.add(this.currentConnection);
         }
 
+        // https://github.com/rudijs/amqp.node-rpc-factory/pull/4
+        // https://github.com/squaremo/amqp.node/issues/110
+        // https://github.com/squaremo/amqp.node/issues/141
+        if (!this.standalone) {
+          conn.on('error', function (err) {
+            this.logError('Publisher: Unexpected amqplib error handled by event:' + err.stack);
+            this.connection = null;
+            this.publisherDomain.remove(this.currentConnection);
+          }.bind(this));
+        }
+
         return conn.createChannel()
           .then(function createChannelSucces(ch) {
 

diff --git a/test/rpc-publisher-factory.spec.js b/test/rpc-publisher-factory.spec.js
@@ -45,11 +45,11 @@ describe('RPC Client', function () {
     done();
   });
 
-  it('should handle emitted AMQP connection errors', function(done) {
+  it('should handle emitted AMQP connection errors', function (done) {
 
     var spy = sinon.spy();
 
-    sinon.stub(domain, 'create', function() {
+    sinon.stub(domain, 'create', function () {
       return new EventEmitter();
     });
 
@@ -300,6 +300,8 @@ describe('RPC Client', function () {
         };
       },
 
+      on: sinon.spy(),
+
       close: sinon.spy()
     };
 
@@ -315,7 +317,7 @@ describe('RPC Client', function () {
       debugLevel: 2
     });
 
-    publisher.standalone.should.be.false;
+    //publisher.standalone.should.be.true;
 
     var spy = sinon.spy(publisher, 'logError');
 
@@ -340,4 +342,64 @@ describe('RPC Client', function () {
 
   });
 
+  it('should handle emitted connection error CONNECTION_FORCED - broker forced connection closure with reason shutdown', function (done) {
+
+    var channelStub = {
+      assertQueue: function () {
+        return {
+          then: function (callback) {
+            callback({queue: uniqueAmqpPrivateReplyQueue});
+            return assertQueueThenStub;
+          }
+        };
+      },
+
+      consume: function (queue, callback, options) {
+        queue.should.equal(uniqueAmqpPrivateReplyQueue);
+        (typeof callback).should.equal('function');
+        options.noAck.should.be.true;
+      },
+
+      sendToQueue: function (replyTo, content, options) {
+        replyTo.should.equal('node_rpc_queue');
+        content.toString().should.equal(messageSent);
+        options.replyTo.should.equal(uniqueAmqpPrivateReplyQueue);
+      },
+
+      close: sinon.spy()
+
+    };
+
+    var createChannelStub = {
+      createChannel: function () {
+        return {
+          then: function (createChannelSuccess) {
+            return createChannelSuccess(channelStub);
+          }
+        };
+      },
+
+      on: function (event, cb) {
+        cb(new Error('CONNECTION_FORCED - broker forced connection closure with reason \'shutdown\''));
+      }
+    };
+
+    var connectStub = {
+      then: function (getConnectionSuccess) {
+        return getConnectionSuccess(createChannelStub);
+      }
+    };
+
+    sinon.stub(amqp, 'connect').returns(connectStub);
+
+    var publisher = rpcPublisherFactory.create({
+      debugLevel: 2
+    });
+
+    publisher.publish(messageSent);
+
+    done();
+
+  });
+
 });