diff --git a/lib/units/device/support/push.js b/lib/units/device/support/push.js index 8ebb40a6..098685ae 100644 --- a/lib/units/device/support/push.js +++ b/lib/units/device/support/push.js @@ -1,8 +1,10 @@ var syrup = require('stf-syrup') var zmq = require('zmq') +var Promise = require('bluebird') var logger = require('../../../util/logger') +var srv = require('../../../util/srv') module.exports = syrup.serial() .define(function(options) { @@ -10,9 +12,14 @@ module.exports = syrup.serial() // Output var push = zmq.socket('push') - options.endpoints.push.forEach(function(endpoint) { - log.info('Sending output to %s', endpoint) - push.connect(endpoint) + Promise.map(options.endpoints.push, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Sending output to "%s"', record.url) + push.connect(record.url) + return Promise.resolve(true) + }) + }) }) return push diff --git a/lib/units/device/support/sub.js b/lib/units/device/support/sub.js index 687b80cd..802cce8a 100644 --- a/lib/units/device/support/sub.js +++ b/lib/units/device/support/sub.js @@ -1,9 +1,11 @@ var syrup = require('stf-syrup') var zmq = require('zmq') +var Promise = require('bluebird') var logger = require('../../../util/logger') var wireutil = require('../../../wire/util') +var srv = require('../../../util/srv') module.exports = syrup.serial() .define(function(options) { @@ -11,9 +13,14 @@ module.exports = syrup.serial() // Input var sub = zmq.socket('sub') - options.endpoints.sub.forEach(function(endpoint) { - log.info('Receiving input from %s', endpoint) - sub.connect(endpoint) + Promise.map(options.endpoints.sub, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Receiving input from "%s"', record.url) + sub.connect(record.url) + return Promise.resolve(true) + }) + }) }) // Establish always-on channels diff --git a/lib/units/notify/hipchat.js b/lib/units/notify/hipchat.js index 79e0f084..71e81d7b 100644 --- a/lib/units/notify/hipchat.js +++ b/lib/units/notify/hipchat.js @@ -9,6 +9,7 @@ var wire = require('../../wire') var wirerouter = require('../../wire/router') var wireutil = require('../../wire/util') var lifecycle = require('../../util/lifecycle') +var srv = require('../../util/srv') var COLORS = { 1: 'gray' @@ -28,9 +29,14 @@ module.exports = function(options) { // Input var sub = zmq.socket('sub') - options.endpoints.sub.forEach(function(endpoint) { - log.info('Receiving input from %s', endpoint) - sub.connect(endpoint) + Promise.map(options.endpoints.sub, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Receiving input from "%s"', record.url) + sub.connect(record.url) + return Promise.resolve(true) + }) + }) }) // Establish always-on channels diff --git a/lib/units/processor/index.js b/lib/units/processor/index.js index b91b1d2c..c11110d6 100644 --- a/lib/units/processor/index.js +++ b/lib/units/processor/index.js @@ -1,3 +1,4 @@ +var Promise = require('bluebird') var zmq = require('zmq') var logger = require('../../util/logger') @@ -6,6 +7,7 @@ var wirerouter = require('../../wire/router') var wireutil = require('../../wire/util') var dbapi = require('../../db/api') var lifecycle = require('../../util/lifecycle') +var srv = require('../../util/srv') module.exports = function(options) { var log = logger.createLogger('processor') @@ -16,9 +18,14 @@ module.exports = function(options) { // App side var appDealer = zmq.socket('dealer') - options.endpoints.appDealer.forEach(function(endpoint) { - log.info('App dealer connected to %s', endpoint) - appDealer.connect(endpoint) + Promise.map(options.endpoints.appDealer, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('App dealer connected to "%s"', record.url) + appDealer.connect(record.url) + return Promise.resolve(true) + }) + }) }) appDealer.on('message', function(channel, data) { @@ -27,9 +34,14 @@ module.exports = function(options) { // Device side var devDealer = zmq.socket('dealer') - options.endpoints.devDealer.forEach(function(endpoint) { - log.info('Device dealer connected to %s', endpoint) - devDealer.connect(endpoint) + Promise.map(options.endpoints.devDealer, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Device dealer connected to "%s"', record.url) + devDealer.connect(record.url) + return Promise.resolve(true) + }) + }) }) devDealer.on('message', wirerouter() diff --git a/lib/units/provider/index.js b/lib/units/provider/index.js index 1a185121..c577fa57 100644 --- a/lib/units/provider/index.js +++ b/lib/units/provider/index.js @@ -11,6 +11,7 @@ var wireutil = require('../../wire/util') var wirerouter = require('../../wire/router') var procutil = require('../../util/procutil') var lifecycle = require('../../util/lifecycle') +var srv = require('../../util/srv') module.exports = function(options) { var log = logger.createLogger('provider') @@ -71,16 +72,26 @@ module.exports = function(options) { // Output var push = zmq.socket('push') - options.endpoints.push.forEach(function(endpoint) { - log.info('Sending output to %s', endpoint) - push.connect(endpoint) + Promise.map(options.endpoints.push, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Sending output to "%s"', record.url) + push.connect(record.url) + return Promise.resolve(true) + }) + }) }) // Input var sub = zmq.socket('sub') - options.endpoints.sub.forEach(function(endpoint) { - log.info('Receiving input from %s', endpoint) - sub.connect(endpoint) + Promise.map(options.endpoints.sub, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Receiving input from "%s"', record.url) + sub.connect(record.url) + return Promise.resolve(true) + }) + }) }) // Establish always-on channels diff --git a/lib/units/reaper/index.js b/lib/units/reaper/index.js index b9561caa..7ee574cc 100644 --- a/lib/units/reaper/index.js +++ b/lib/units/reaper/index.js @@ -6,6 +6,7 @@ var wire = require('../../wire') var wireutil = require('../../wire/util') var dbapi = require('../../db/api') var lifecycle = require('../../util/lifecycle') +var srv = require('../../util/srv') module.exports = function(options) { var log = logger.createLogger('reaper') @@ -17,9 +18,14 @@ module.exports = function(options) { // Output var push = zmq.socket('push') - options.endpoints.push.forEach(function(endpoint) { - log.info('Sending output to %s', endpoint) - push.connect(endpoint) + Promise.map(options.endpoints.push, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Sending output to "%s"', record.url) + push.connect(record.url) + return Promise.resolve(true) + }) + }) }) function reap() { diff --git a/lib/units/websocket/index.js b/lib/units/websocket/index.js index 38617197..8484952e 100644 --- a/lib/units/websocket/index.js +++ b/lib/units/websocket/index.js @@ -15,6 +15,7 @@ var wireutil = require('../../wire/util') var wirerouter = require('../../wire/router') var dbapi = require('../../db/api') var datautil = require('../../util/datautil') +var srv = require('../../util/srv') var cookieSession = require('./middleware/cookie-session') var ip = require('./middleware/remote-ip') var auth = require('./middleware/auth') @@ -30,16 +31,26 @@ module.exports = function(options) { // Output var push = zmq.socket('push') - options.endpoints.push.forEach(function(endpoint) { - log.info('Sending output to %s', endpoint) - push.connect(endpoint) + Promise.map(options.endpoints.push, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Sending output to "%s"', record.url) + push.connect(record.url) + return Promise.resolve(true) + }) + }) }) // Input var sub = zmq.socket('sub') - options.endpoints.sub.forEach(function(endpoint) { - log.info('Receiving input from %s', endpoint) - sub.connect(endpoint) + Promise.map(options.endpoints.sub, function(endpoint) { + return srv.resolve(endpoint).then(function(records) { + return srv.attempt(records, function(record) { + log.info('Receiving input from "%s"', record.url) + sub.connect(record.url) + return Promise.resolve(true) + }) + }) }) // Establish always-on channels diff --git a/lib/util/srv.js b/lib/util/srv.js index cd49379f..7bb2bd68 100644 --- a/lib/util/srv.js +++ b/lib/util/srv.js @@ -4,6 +4,8 @@ var util = require('util') var Promise = require('bluebird') var dns = Promise.promisifyAll(require('dns')) +var srv = module.exports = Object.create(null) + function groupByPriority(records) { function sortByPriority(a, b) { return a.priority - b.priority @@ -61,11 +63,21 @@ function flatten(groupedRecords) { }, []) } -module.exports.sort = function(records) { +function NEXT() { + Error.call(this) + this.name = 'NEXT' + Error.captureStackTrace(this, NEXT) +} + +util.inherits(NEXT, Error) + +srv.NEXT = NEXT + +srv.sort = function(records) { return flatten(groupByPriority(records).map(shuffleWeighted)) } -module.exports.resolve = function(domain) { +srv.resolve = function(domain) { var parsedUrl = url.parse(domain) if (!parsedUrl.protocol) { @@ -97,3 +109,17 @@ module.exports.resolve = function(domain) { }]) } } + +srv.attempt = function(records, fn) { + function next(i) { + if (i >= records.length) { + throw new Error('No more records left to try') + } + + return fn(records[i]).catch(srv.NEXT, function() { + return next(++i) + }) + } + + return next(0) +}