src/exchanges.js
function(entries, exchangePrefix, connectionFunc, options) {
  events.EventEmitter.call(this);
  assert(options.validator, 'options.validator must be provided');
  this._conn = null;
  this.__reconnectTimer = null;
  this._connectionFunc = connectionFunc;
  this._channel = null;
  this._connecting = null;
  this._entries = entries;
  this._exchangePrefix = exchangePrefix;
  this._options = options;
  this._errCount = 0;
  this._lastErr = Date.now();
  this._lastTime = 0;
  this._sleeping = null;
  this._sleepingTimeout = null;
  if (options.drain || options.component) {
    console.log('taskcluster-lib-stats is now deprecated!\n' +
                'Use the `monitor` option rather than `drain`.\n' +
                '`monitor` should be an instance of taskcluster-lib-monitor.\n' +
                '`component` is no longer needed. Prefix your `monitor` before use.');
  }

  var monitor = null;
  if (options.monitor) {
    monitor = options.monitor;
  }

  entries.forEach((entry) => {
    this[entry.name] = (...args) => {
      // Construct message and routing key from arguments
      var message = entry.messageBuilder.apply(undefined, args);
      common.validateMessage(this._options.rootUrl, this._options.serviceName, this._options.version,
        options.validator, entry, message);

      var routingKey = common.routingKeyToString(entry, entry.routingKeyBuilder.apply(undefined, args));

      var CCs = entry.CCBuilder.apply(undefined, args);
      assert(CCs instanceof Array, 'CCBuilder must return an array');

      // Serialize message to buffer
      var payload = new Buffer(JSON.stringify(message), 'utf8');

      // Find exchange name
      var exchange = exchangePrefix + entry.exchange;

      // Log that we're publishing a message
      debug('Publishing message on exchange: %s', exchange);

      // Return promise
      return this._connect().then(channel => {
        return new Promise((accept, reject) => {
          // Start timer
          var start = null;
          if (monitor) {
            start = process.hrtime();
          }

          // Set a timeout
          let done = false;
          this._sleep12Seconds().then(() => {
            if (!done) {
              let err = new Error('publish message timed out after 12s');
              this._handleError(err);
              reject(err);
            }
          });

          // Publish message
          channel.publish(exchange, routingKey, payload, {
            persistent:         true,
            contentType:        'application/json',
            contentEncoding:    'utf-8',
            CC:                 CCs,
          }, (err, val) => {
            // NOTE: many channel errors will not invoke this callback at all,
            // hence the 12-second timeout
            done = true;
            if (monitor) {
              var d = process.hrtime(start);
              monitor.measure(exchange, d[0] * 1000 + d[1] / 1000000);
              monitor.count(exchange);
            }

            // Handle errors
            if (err) {
              err.methodName = entry.name;
              err.exchange = exchange;
              err.routingKey = routingKey;
              err.payload = payload;
              err.ccRoutingKeys = CCs;
              debug('Failed to publish message: %j and routingKey: %s, ' +
              'with error: %s, %j', message, routingKey, err, err);
              if (monitor) {
                monitor.reportError(err);
              }
              return reject(err);
            }
            accept(val);
          });
        });
      });
    };
  });
}
function(entries, exchangePrefix, connectionFunc, options) {
  events.EventEmitter.call(this);
  assert(options.validator, 'options.validator must be provided');
  this._conn = null;
  this.__reconnectTimer = null;
  this._connectionFunc = connectionFunc;
  this._channel = null;
  this._connecting = null;
  this._entries = entries;
  this._exchangePrefix = exchangePrefix;
  this._options = options;
  this._errCount = 0;
  this._lastErr = Date.now();
  this._lastTime = 0;
  this._sleeping = null;
  this._sleepingTimeout = null;
  if (options.drain || options.component) {
    console.log('taskcluster-lib-stats is now deprecated!\n' +
                'Use the `monitor` option rather than `drain`.\n' +
                '`monitor` should be an instance of taskcluster-lib-monitor.\n' +
                '`component` is no longer needed. Prefix your `monitor` before use.');
  }

  var monitor = null;
  if (options.monitor) {
    monitor = options.monitor;
  }

  entries.forEach((entry) => {
    this[entry.name] = (...args) => {
      // Construct message and routing key from arguments
      var message = entry.messageBuilder.apply(undefined, args);
      common.validateMessage(this._options.rootUrl, this._options.serviceName, this._options.version,
        options.validator, entry, message);

      var routingKey = common.routingKeyToString(entry, entry.routingKeyBuilder.apply(undefined, args));

      var CCs = entry.CCBuilder.apply(undefined, args);
      assert(CCs instanceof Array, 'CCBuilder must return an array');

      // Serialize message to buffer
      var payload = new Buffer(JSON.stringify(message), 'utf8');

      // Find exchange name
      var exchange = exchangePrefix + entry.exchange;

      // Log that we're publishing a message
      debug('Publishing message on exchange: %s', exchange);

      // Return promise
      return this._connect().then(channel => {
        return new Promise((accept, reject) => {
          // Start timer
          var start = null;
          if (monitor) {
            start = process.hrtime();
          }

          // Set a timeout
          let done = false;
          this._sleep12Seconds().then(() => {
            if (!done) {
              let err = new Error('publish message timed out after 12s');
              this._handleError(err);
              reject(err);
            }
          });

          // Publish message
          channel.publish(exchange, routingKey, payload, {
            persistent:         true,
            contentType:        'application/json',
            contentEncoding:    'utf-8',
            CC:                 CCs,
          }, (err, val) => {
            // NOTE: many channel errors will not invoke this callback at all,
            // hence the 12-second timeout
            done = true;
            if (monitor) {
              var d = process.hrtime(start);
              monitor.measure(exchange, d[0] * 1000 + d[1] / 1000000);
              monitor.count(exchange);
            }

            // Handle errors
            if (err) {
              err.methodName = entry.name;
              err.exchange = exchange;
              err.routingKey = routingKey;
              err.payload = payload;
              err.ccRoutingKeys = CCs;
              debug('Failed to publish message: %j and routingKey: %s, ' +
              'with error: %s, %j', message, routingKey, err, err);
              if (monitor) {
                monitor.reportError(err);
              }
              return reject(err);
            }
            accept(val);
          });
        });
      });
    };
  });
}
function ( entries , exchangePrefix , connectionFunc , options ) { events . EventEmitter . call ( this ) ; assert ( options . validator , 'options.validator must be provided' ) ; this . _conn = null ; this . __reconnectTimer = null ; this . _connectionFunc = connectionFunc ; this . _channel = null ; this . _connecting = null ; this . _entries = entries ; this . _exchangePrefix = exchangePrefix ; this . _options = options ; this . _errCount = 0 ; this . _lastErr = Date . now ( ) ; this . _lastTime = 0 ; this . _sleeping = null ; this . _sleepingTimeout = null ; if ( options . drain || options . component ) { console . log ( 'taskcluster-lib-stats is now deprecated!\n' + 'Use the `monitor` option rather than `drain`.\n' + '`monitor` should be an instance of taskcluster-lib-monitor.\n' + '`component` is no longer needed. Prefix your `monitor` before use.' ) ; } var monitor = null ; if ( options . monitor ) { monitor = options . monitor ; } entries . forEach ( ( entry ) => { this [ entry . name ] = ( ... args ) => { // Construct message and routing key from arguments var message = entry . messageBuilder . apply ( undefined , args ) ; common . validateMessage ( this . _options . rootUrl , this . _options . serviceName , this . _options . version , options . validator , entry , message ) ; var routingKey = common . routingKeyToString ( entry , entry . routingKeyBuilder . apply ( undefined , args ) ) ; var CCs = entry . CCBuilder . apply ( undefined , args ) ; assert ( CCs instanceof Array , 'CCBuilder must return an array' ) ; // Serialize message to buffer var payload = new Buffer ( JSON . stringify ( message ) , 'utf8' ) ; // Find exchange name var exchange = exchangePrefix + entry . exchange ; // Log that we're publishing a message debug ( 'Publishing message on exchange: %s' , exchange ) ; // Return promise return this . _connect ( ) . then ( channel => { return new Promise ( ( accept , reject ) => { // Start timer var start = null ; if ( monitor ) { start = process . hrtime ( ) ; } // Set a timeout let done = false ; this . _sleep12Seconds ( ) . then ( ( ) => { if ( ! done ) { let err = new Error ( 'publish message timed out after 12s' ) ; this . _handleError ( err ) ; reject ( err ) ; } } ) ; // Publish message channel . publish ( exchange , routingKey , payload , { persistent : true , contentType : 'application/json' , contentEncoding : 'utf-8' , CC : CCs , } , ( err , val ) => { // NOTE: many channel errors will not invoke this callback at all, // hence the 12-second timeout done = true ; if ( monitor ) { var d = process . hrtime ( start ) ; monitor . measure ( exchange , d [ 0 ] * 1000 + d [ 1 ] / 1000000 ) ; monitor . count ( exchange ) ; } // Handle errors if ( err ) { err . methodName = entry . name ; err . exchange = exchange ; err . routingKey = routingKey ; err . payload = payload ; err . ccRoutingKeys = CCs ; debug ( 'Failed to publish message: %j and routingKey: %s, ' + 'with error: %s, %j' , message , routingKey , err , err ) ; if ( monitor ) { monitor . reportError ( err ) ; } return reject ( err ) ; } accept ( val ) ; } ) ; } ) ; } ) ; } ; } ) ; }
Class for publishing to a set of declared exchanges
Class for publishing to a set of declared exchanges