harcon-amqp - npm Package Compare versions

Comparing version 3.1.0 to 3.2.0

lib/AmqpBarrel.js

@@ -45,2 +45,3 @@ 'use strict'
 	socket.on('error', self.logger.error )
+	socket.on('close', self.logger.error )
 }
@@ -66,2 +67,3 @@
 	socket.on('error', self.logger.error )
+	socket.on('close', self.logger.error )
 }
@@ -73,2 +75,4 @@
 	self.messages = {}
+	self.outs = {}
+	self.ins = {}
 
@@ -83,18 +87,7 @@ self.nodeSeqNo = config.nodeSeqNo || 1
 	self.expiration = config.expiration || 0
-	self.ctx = rabbit.createContext( self.connectURL )
-	self.ctx.on('ready', function () {
-		self.logger.harconlog( null, 'AMQP connection is made.', self.connectURL, 'info' )
-		self.outs = {}
-		self.ins = {}
 
-		if ( callback )
-			callback()
-	} )
-	self.ctx.on('error', self.logger.error )
+	self.reconnectionTimeout = config.reconnectionTimeout || 500
+	self.reconnectionMaxTimeout = config.reconnectionMaxTimeout || 10000
 
-	if ( self.timeout > 0 ) {
-		self.cleaner = setInterval( function () {
-			self.cleanupMessages()
-		}, self.timeout )
-	}
+	self.connect( callback )
 }
@@ -212,9 +205,72 @@
 
+amqpbarrel.clearClearer = function ( ) {
+	if ( this.cleaner ) {
+		clearInterval( this.cleaner )
+		this.cleaner = null
+	}
+}
+
+amqpbarrel.connect = function ( callback ) {
+	let self = this
+
+	self.ctx = rabbit.createContext( self.connectURL )
+	self.reconnectionProcess = null
+
+	self.ctx.on('error', self.reconnect.bind( self ) )
+	self.ctx.on('close', self.reconnect.bind( self ) )
+	self.ctx.on('ready', function () {
+		self.logger.harconlog( null, 'AMQP connection is made.', self.connectURL, 'warn' )
+
+		self.setupDomains( function () {
+			if ( callback )
+				callback()
+		} )
+	} )
+
+	self.clearClearer()
+	if ( self.timeout > 0 ) {
+		self.cleaner = setInterval( function () {
+			self.cleanupMessages()
+		}, self.timeout )
+	}
+}
+
+amqpbarrel.reconnect = function ( ) {
+	let self = this
+
+	if (self.reconnectionProcess) return
+
+	self.logger.harconlog( null, 'Reconnecting...', self.connectURL, 'warn' )
+
+	self.reconnectionProcess = setTimeout( function () {
+		self.connect( function () {
+			self.reconnectionTimeout *= 2
+			if ( self.reconnectionTimeout > self.reconnectionMaxTimeout )
+				self.reconnectionTimeout = self.reconnectionMaxTimeout
+		} )
+	}, self.reconnectionTimeout )
+}
+
+amqpbarrel.setupDomains = function ( callback ) {
+	let self = this
+
+	let fns = []
+	Object.keys(self.ins).forEach( function (domain) {
+		Object.keys(self.ins[domain]).forEach( function (entity) {
+			fns.push( function (cb) { self.createIn( domain, entity, cb ) } )
+		} )
+	} )
+	Object.keys(self.outs).forEach( function (division) {
+		fns.push( function (cb) { self.createOut( division, cb ) } )
+	} )
+	async.series( fns, callback )
+}
+
 amqpbarrel.extendedClose = function ( callback ) {
-	if ( this.cleaner )
-		clearInterval( this.cleaner )
-	if ( this.ctx )
+	this.clearClearer()
+	if ( this.ctx ) {
 		this.ctx.close( callback )
+	}
 }
 
 module.exports = AmqpBarrel

package.json

 {
   "name": "harcon-amqp",
-  "version": "3.1.0",
+  "version": "3.2.0",
   "description": "AMQP plugin for the harcon messaging/service bus of node-based enterprise entities.",
@@ -5,0 +5,0 @@ "keywords": [

test/amqp.mocha.js

@@ -25,3 +25,3 @@ 'use strict'
 	before(function (done) {
-		var logger = Logger.createWinstonLogger( { file: 'mochatest.log', level: 'debug' } )
+		var logger = Logger.createWinstonLogger( { console: true, /* file: 'mochatest.log', */ level: 'debug' } )
 
@@ -70,2 +70,24 @@ // Initializes the Harcon system
 		})
+		/* it('test network failure', function (done) {
+			console.log('test network failure...')
+			this.timeout(20000)
+			setTimeout( function () {
+				inflicter.ignite( clerobee.generate(), null, '', 'Marie.simple', 'whatsup?', 'how do you do?', function (err, res) {
+					console.log('????????', err, res)
+				})
+			}, 7000)
+			setTimeout( function () {
+				console.log('SENDING!!!!!!!')
+				inflicter.ignite( clerobee.generate(), null, '', 'Marie.simple', 'whatsup?', 'how do you do?', function (err, res) {
+					console.log('>>>>>', err, res)
+					should.not.exist(err)
+					should.exist(res)
+
+					expect( res ).to.include( 'Bonjour!' )
+
+					done()
+				})
+			}, 15000)
+		})*/
+
 		it('Retrieve listeners...', function (done) {
@@ -77,12 +99,2 @@ inflicter.listeners( function (err, listeners) {
 		})
-		/*
-		it('Send for divisions...', function (done) {
-			inflicter.ignite( clerobee.generate(), '', 'Inflicter.divisions', function (err, res) {
-				should.not.exist(err)
-				should.exist(res)
-				expect( res[0] ).to.include( harconName, harconName + '.click' )
-				done()
-			} )
-		})
-		*/
 		it('Clean internals', function (done) {
@@ -96,2 +108,3 @@ inflicter.pendingComms( function (err, comms) {
 		})
+
 	})
@@ -249,3 +262,2 @@
 		})
-
 	})
@@ -252,0 +264,0 @@

test/AMQPQueueTest.js

@@ -5,5 +5,4 @@ 'use strict'
 
-let ctx = rabbit.createContext( 'amqp://localhost' )
+let ctx
 
-
 let julie = {}
@@ -17,2 +16,8 @@ function createJulie ( division, name ) {
 	} )
+	julie.pub.on('close', function () {
+		console.error('>>>>> julie pub >>>>>>', arguments)
+	} )
+	julie.sub.on('close', function () {
+		console.error('>>>>> julie sub >>>>>>', arguments)
+	} )
 	julie.sub.connect( division, 'julie.*', function (err) {
@@ -36,2 +41,5 @@ if (err) console.error(err)
 	} )
+	marie.sub.on('close', function () {
+		console.error('>>>>> marie sub >>>>>>', arguments)
+	} )
 	marie.sub.connect( division, 'marie.*', function ( ) {
@@ -54,6 +62,14 @@ console.log('Marie sub connected')
 
-ctx.on('ready', function () {
-	console.log('Connected')
-	connected()
+function create ( closeFn ) {
+	ctx = rabbit.createContext( 'amqp://localhost' )
+	ctx.on('ready', function () {
+		console.log('Connected')
+		connected()
+	} )
+	ctx.on('error', closeFn )
+	ctx.on('close', closeFn )
+}
+
+create( function () {
+	console.error('>>>>> WTF >>>>>>', arguments)
 } )
-ctx.on('error', console.error )

No alert changes

Improved metrics

Total package byte prevSize

33423

increased by7.31%

Lines of code

887

increased by8.44%

No dependency changes