/*
! @param {AkeraConnector} Akera connector class
427651
function DiscoveryDecorator(AkeraConnector) {

  function getDatabasesMeta(conn, cb) {
    conn.getMetaData(function(err, meta) {
      if (err !== null)
        cb(err);
      else {
        meta.allDatabases().then(function(dbs) {
          cb(null, dbs);
        }, cb);
      }
    });
  }

  function getTableMeta(conn, schema, table, cb) {
    getDatabasesMeta(conn,
        function(err, dbs) {
          if (err !== null)
            cb(err);
          else {
            var callback = function(err) {
              if (err !== null)
                cb(err);

              for ( var t in tables) {
                if (table === tables[t].getName()) {
                  cb(null, tables[t]);
                  return;
                }
              }

              cb(new Error('Table not found in database: ' + schema + '.'
                  + table));
            };

            // select first database as default if not specified
            if (!schema)
              schema = dbs[0].getLname();

            for ( var d in dbs) {
              if (schema === dbs[d].getLname()) {
                var tables = [];
                var info = {
                  idx : d,
                  schema : schema
                };

                fetchSchemaTables(dbs, info, tables, callback);

                return;
              }
            }
            cb(new Error('Database not connected: ' + schema));
          }
        });
  }

  var dataTypeConversion = {
    DATE : 'Date',
    DATETIME : 'Date',
    'DATETIME-TZ' : 'Date',
    LOGICAL : 'Boolean',
    DECIMAL : 'Number',
    INTEGER : 'Number',
    INT64 : 'Number',
    CHARACTER : 'String',
    CLOB : 'String',
    BLOB : 'Buffer'
  };

  function getTableModel(tblMeta, cb) {
    tblMeta.getAllFields().then(function(fields) {
      var result = [];

      for ( var f in fields) {
        var field = fields[f];

        result.push({
          owner : tblMeta.getDatabase(),
          tableName : tblMeta.getName(),
          columnName : field.name,
          dataType : field.type,
          nullable : field.mandatory ? 'N' : 'Y',
          type : dataTypeConversion[field.type.toUpperCase()]
        });
      }

      cb(null, result);
    }, cb);
  }

  function getTablePrimaryKeys(tblMeta, cb) {
    tblMeta.getPk().then(function(pk) {
      var result = [];

      for ( var f in pk.fields) {
        var field = pk.fields[f];

        result.push({
          owner : tblMeta.getDatabase(),
          tableName : tblMeta.getName(),
          columnName : field.name,
          keySeq : field.fld_pos,
          pkName : field.idx_name
        });
      }

      cb(null, result);
    }, cb);
  }

  function getDatabaseSchemas(dbs, options) {
    var offset = 0;
    var limit = dbs.length;
    var result = [];

    if (options !== null && typeof options === 'object') {
      if (typeof options.offset === 'number')
        offset = options.offset;
      if (typeof options.skip === 'number')
        offset += options.skip;
      if (typeof options.limit === 'number')
        limit = options.limit;
    }

    for (var i = offset; i < limit; i++) {
      result.push({
        catalog : dbs[i].getLname(),
        schema : dbs[i].getLname()
      });
    }

    return result;
  }

  function getModelDefinitions(tblsMeta) {
    var result = [];

    for ( var t in tblsMeta) {
      result.push({
        type : 'table',
        name : tblsMeta[t].getName(),
        owner : tblsMeta[t].getDatabase()
      });
    }

    return result;
  }

  function getDatabaseTables(dbMeta, cb) {
    dbMeta.allTables().then(function(tables) {
      cb(null, tables);
    }, cb);
  }

  function fetchSchemaTables(dbs, info, tables, cb) {
    getDatabaseTables(dbs[info.idx], function(err, tbls) {
      if (err !== null)
        cb(err);
      else {
        for ( var t in tbls) {
          tables.push(tbls[t]);
        }
        // reach the end of databases or found the one we're looking for
        if ((info.schema && info.schema === dbs[info.idx].getLname())
            || ++info.idx === dbs.length)
          cb(null, tables);
        else
          fetchSchemaTables(dbs, info, tables, cb);
      }
    });
  }

  AkeraConnector.prototype.discoverDatabaseSchemas = function(options, cb) {
    if (!cb && typeof options === 'function') {
      cb = options;
      options = null;
    }

    getDatabasesMeta(this, function(err, dbs) {
      if (err !== null)
        cb(err);
      else {
        cb(null, getDatabaseSchemas(dbs, options));
      }
    });
  };

  /**
   * Discover model definitions
   * 
   * @param {Object}
   *         options Options for discovery
   * @param {Function}
   *         [cb] The callback function
   */
  AkeraConnector.prototype.discoverModelDefinitions = function(options, cb) {
    if (!cb && typeof options === 'function') {
      cb = options;
      options = null;
    }

    var schema = null;
    if (options !== null && typeof options === 'object') {
      schema = options.owner || options.schema;
    }

    getDatabasesMeta(this, function(err, dbs) {
      if (err !== null)
        cb(err);
      else {
        var tables = [];
        var info = {
          idx : 0,
          schema : schema
        };

        fetchSchemaTables(dbs, info, tables, function(err) {
          if (err !== null)
            cb(err);

          cb(null, getModelDefinitions(tables));
        });
      }
    });

  };

  /**
   * Discover model properties from a table
   * 
   * @param {String}
   *         table The table name
   * @param {Object}
   *         options The options for discovery (schema/owner)
   * @param {Function}
   *         [cb] The callback function
   * 
   */
  AkeraConnector.prototype.discoverModelProperties = function(table, options,
      cb) {
    var schema = null;

    if (!cb && typeof options === 'function') {
      cb = options;
      options = null;
    }

    if (table === null && typeof table !== 'string') {
      cb(new Error('Table name is mandatory for model properties discovery.'));
      return;
    }

    if (options !== null && typeof options === 'object') {
      schema = options.owner || options.schema;
    }

    if (!schema) {
      schema = this.getDefaultSchema();
    }

    getTableMeta(this, schema, table, function(err, tblMeta) {
      if (err) {
        cb(err);
      } else {
        getTableModel(tblMeta, cb);
      }
    });
  };

  /**
   * Discover model primary keys from a table
   * 
   * @param {String}
   *         table The table name
   * @param {Object}
   *         options The options for discovery (schema/owner)
   * @param {Function}
   *         [cb] The callback function
   * 
   */
  AkeraConnector.prototype.discoverPrimaryKeys = function(table, options, cb) {
    var schema = null;

    if (!cb && typeof options === 'function') {
      cb = options;
      options = null;
    }

    if (table === null && typeof table !== 'string') {
      cb(new Error('Table name is mandatory for model primary keys discovery.'));
      return;
    }

    if (options !== null && typeof options === 'object') {
      schema = options.owner || options.schema;
    }

    if (!schema) {
      schema = this.getDefaultSchema();
    }

    getTableMeta(this, schema, table, function(err, tblMeta) {
      if (err) {
        cb(err);
      } else {
        getTablePrimaryKeys(tblMeta, cb);
      }
    });
  };

  AkeraConnector.prototype.getDefaultSchema = function() {
    if (this.dataSource && this.dataSource.settings
        && this.dataSource.settings.database) {
      return this.dataSource.settings.database;
    }
    return undefined;
  };
}
301441
function attach(listener) {
    if (!listener) return;
    if (!listener[CONTEXTS_SYMBOL]) listener[CONTEXTS_SYMBOL] = Object.create(null);

    listener[CONTEXTS_SYMBOL][thisSymbol] = {
      namespace : namespace,
      context   : namespace.active
    };
  }
250462
function ModulesCtrl($scope, $location, $rootScope, $window, $timeout, $sce, 
                     _bind, _modules, _req, _socket) {

  /****************************************************************************/
  /* INITIALIZATION                                                           */
  /****************************************************************************/
  /* Handhsaking [modules] */
  _socket.on('modules', function(state) {
    //console.log('========================================');
    //console.log(JSON.stringify(state, null, 2));
    //console.log('----------------------------------------');
    $scope.modules = state;
  });

  /* Handhsaking [about] */
  _socket.on('about', function(state) {
    //console.log('========================================');
    //console.log(JSON.stringify(state, null, 2));
    //console.log('----------------------------------------');
    $scope.about = state;
  });

  _socket.emit('handshake', 'modules');
  _socket.emit('handshake', 'about');

  $window.document.title = 'Breach::Modules';

  /****************************************************************************/
  /* COMMANDS                                                                  */
  /****************************************************************************/
  $scope.modules_install = function() {
    var path = $scope.install_path;
    $scope.install_path = '';
    async.waterfall([
      function(cb_) {
        _modules.add(path).then(function(data) {
          return cb_(null, data.module);
        });
      },
      function(module, cb_) {
        _modules.install(module.path).then(function(data) {
          return cb_(null, data.module);
        });
      },
      function(module, cb_) {
        _modules.run(module.path).then(function(data) {
          return cb_(null, data.module);
        });
      }
    ], function(err) {
    });
  };

  $scope.modules_remove = function(path) {
    _modules.remove(path).then(function(data) {
    });
  };

  $scope.modules_update = function(path) {
    _modules.update(path).then(function(data) {
    });
  };

  $scope.modules_kill = function(path) {
    _modules.kill(path).then(function(data) {
    });
  };
  $scope.modules_run = function(path) {
    _modules.run(path).then(function(data) {
    });
  };

  $scope.modules_restart = function(path) {
    async.series([
      function(cb_) {
        _modules.kill(path).then(function(data) {
          return cb_();
        });
      },
      function(cb_) {
        _modules.run(path).then(function(data) {
          return cb_();
        });
      },
    ], function(err) {
    });
  };

  $scope.about_install = function() {
    _req.post('/about/install', {}).then(function(data) {
    });
  };
}
727507
function wrap() {
  var ports = this.compiledModule.ports;

  var incomingEmitter = new EventEmitter();
  var outgoingEmitter = new EventEmitter();

  var emit = incomingEmitter.emit.bind(incomingEmitter);

  Object.keys(ports).forEach(function(key) {
    outgoingEmitter.addListener(key, function() {
      var args = Array.prototype.slice.call(arguments)

      ports[key].send.apply(ports[key], args);
    });

    if (ports[key].subscribe) {
      ports[key].subscribe(function() {
        var args = Array.prototype.slice.call(arguments);
        args.unshift(key);

        emit.apply(incomingEmitter, args);
      });
    }
  });

  incomingEmitter.emit = outgoingEmitter.emit.bind(outgoingEmitter);;

  this.emitter = incomingEmitter;
  this.ports = this.compiledModule.ports;
}
402230
async function initModule (ravelInstance) {
  const self = this;
  const name = Metadata.getClassMetaValue(Object.getPrototypeOf(self), '@role', 'name');
  // run @postinject handlers
  const postInjects = Metadata.getClassMeta(Object.getPrototypeOf(self), '@postinject', Object.create(null));
  await Promise.all(Object.keys(postInjects).map(f => postInjects[f].call(self)));
  // connect any Ravel lifecycle handlers to the appropriate events
  function connectHandlers (decorator, event) {
    const handlers = Metadata.getClassMeta(Object.getPrototypeOf(self), decorator, Object.create(null));
    for (const f of Object.keys(handlers)) {
      ravelInstance.once(event, (...args) => {
        ravelInstance.$log.trace(`${name}: Invoking ${decorator} ${f}`);
        handlers[f].apply(self, args);
      });
    }
  }
  connectHandlers('@postinit', 'post init');
  connectHandlers('@prelisten', 'pre listen');
  connectHandlers('@koaconfig', 'pre routes init');
  connectHandlers('@postlisten', 'post listen');
  connectHandlers('@preclose', 'end');

  // connect @interval
  const handlers = Metadata.getClassMeta(Object.getPrototypeOf(self), '@interval', Object.create(null));
  const intervals = [];
  ravelInstance.once('post listen', () => {
    for (const f of Object.keys(handlers)) {
      intervals.push(setInterval(() => {
        ravelInstance.$log.trace(`${name}: Invoking @interval ${f}`);
        handlers[f].handler.call(self);
      }, handlers[f].interval));
    }
  });
  ravelInstance.once('end', () => {
    intervals.forEach((i) => clearInterval(i));
  });

  // register middleware
  const middleware = Metadata.getClassMeta(Object.getPrototypeOf(self), '@middleware', Object.create(null));
  ravelInstance.once('post module init', () => {
    for (const m of Object.keys(middleware)) {
      if (ravelInstance[symbols.moduleFactories][m]) {
        throw new $err.DuplicateEntry(
          `Unable to register @middleware with name ${m}, which conflicts with an existing Module name`);
      } else {
        ravelInstance.$log.info(`Registering middleware with name ${m}`);
        ravelInstance[symbols.middleware][m] = middleware[m].bind(self);
      }
    }
  });
}
1031511
function use(adapter) {
  if (~adapters.indexOf(adapter)) return false;

  adapters.push(adapter);
  return true;
}
1214662
function Connection (options) {
  options = utils.clone(options || {});

  var ns = this._ns = utils.createNamespace(options);
  var id = this.id = options.id || utils.uniqueId('connection');
  this._debug = utils.debug('connection:' + id);
  var me = this;

  // create controller connection
  this._host = options.host || define.host;
  this._port = options.port || define.port;
  process.nextTick(function () {
    me._debug('trying reconnect');
    me._reconnect();
  });

  me._queue = [];
  ['send', 'registerKey', 'deleteKey', 'deleteKeys', 'keys'].forEach(function (method) {
    me[method] = me._wrapMethod(me[method], method);
  });
  me.on('ready', function () {
    me._debug('emit listen');
    me.emit('listen');
    me._processQueue();
  });
}
543002
function Emitter(conn) {
            if (!(this instanceof Emitter)) return new Emitter(conn);
            this.ids = 1;
            this.acks = {};
            this.conn = conn;
            if (this.conn) this.bind();
          }
620967
function ConnectionManager(namespace) {
  this.namespace = namespace;
  this.clients = new ArrayKeys({
    identifier: 'id'
  });

  var reconnecting = false;
  var self = this;

  if (! reportingActive) {
    reportingActive = true;

    setInterval(function __report() {
      self.clients.forEachRecord(function (client) {
        var count = client.references.getCount();

        if ((count > 0) && ((typeof client.isConnected !== 'function') || (! client.isConnected())) && (! reconnecting)) {
          debug('client ' + client.id + ' is disconnected, but reference count: ' + count + ', reconnecting...');

          var scopes   = client.references.getIdentifiers();
          var scope    = client.references.getRecord(scopes[0]);
          reconnecting = true;

          client.connect.call(buildScope(scope, client), function (err, c) {
            if (err) {
              debug('failed reconnecting: ', err);
              throw new Error(err);
            }
            debug('client successfully reconnected.');
            client.connection = c;
            client = registerListeners(scope, client);
            reconnecting = false;
          });

          for (var i = 1, len = scopes.length; i < len; i++) {
            client = registerListeners(client.references.getRecord(scopes[i]), client);
          }
        }
      });
    }, 60000);
  }
}
163494
function Connection(options) {
  //*****************************************************************
  // public API functions
  //*****************************************************************

  /**
   * Connect event
   *
   * @returns {Promise} promise
   */
  this.connect = () => {
    switch (_status) {
      case Status.NOT_CONNECTED:
        _status = Status.CONNECTING;
        return new Promise(function(resolve, reject) {
          _registerHandshakeCmd(resolve, reject);
        });

      case Status.CLOSING:
      case Status.CLOSED:
        return Promise.reject(
          Errors.createError(
            'Connection closed',
            true,
            info,
            '08S01',
            Errors.ER_CONNECTION_ALREADY_CLOSED
          )
        );

      case Status.CONNECTING:
      case Status.AUTHENTICATING:
        return Promise.reject(
          Errors.createError(
            'Connection is already connecting',
            true,
            info,
            '08S01',
            Errors.ER_ALREADY_CONNECTING
          )
        );
    }
    //status Connected
    return Promise.resolve(this);
  };

  /**
   * Permit to change user during connection.
   * All user variables will be reset, Prepare commands will be released.
   * !!! mysql has a bug when CONNECT_ATTRS capability is set, that is default !!!!
   *
   * @param options   connection options
   * @returns {Promise} promise
   */
  this.changeUser = options => {
    if (!info.isMariaDB()) {
      return Promise.reject(
        Errors.createError(
          'method changeUser not available for MySQL server due to Bug #83472',
          false,
          info,
          '0A000',
          Errors.ER_MYSQL_CHANGE_USER_BUG
        )
      );
    }

    return new Promise(function(resolve, reject) {
      _addCommand(
        new ChangeUser(
          options,
          resolve,
          _authFailHandler.bind(this, reject),
          _addCommand.bind(this)
        )
      );
    });
  };

  /**
   * Start transaction
   *
   * @returns {Promise} promise
   */
  this.beginTransaction = () => {
    return this.query('START TRANSACTION');
  };

  /**
   * Commit a transaction.
   *
   * @returns {Promise} command if commit was needed only
   */
  this.commit = () => {
    return _changeTransaction('COMMIT');
  };

  /**
   * Roll back a transaction.
   *
   * @returns {Promise} promise
   */
  this.rollback = () => {
    return _changeTransaction('ROLLBACK');
  };

  /**
   * Execute query using text protocol.
   *
   * @param sql     sql parameter Object can be used to supersede default option.
   *                Object must then have sql property.
   * @param values  object / array of placeholder values (not mandatory)
   * @returns {Promise} promise
   */
  this.query = (sql, values) => {
    let _cmdOpt, _sql;
    if (typeof sql === 'object') {
      _cmdOpt = sql;
      _sql = _cmdOpt.sql;
    } else {
      _sql = sql;
    }

    return new Promise(function(resolve, reject) {
      const cmd = new Query(resolve, reject, _cmdOpt, opts, _sql, values);
      if (opts.trace) Error.captureStackTrace(cmd);
      _addCommand(cmd);
    });
  };

  /**
   * Execute batch using text protocol.
   *
   * @param sql     sql parameter Object can be used to supersede default option.
   *                Object must then have sql property.
   * @param initialValues  object / array of placeholder values (not mandatory)
   * @returns {Promise} promise
   */
  this.batch = (sql, initialValues) => {
    let _options, _sql;
    if (typeof sql === 'object') {
      _options = sql;
      _sql = _options.sql;
    } else {
      _sql = sql;
    }

    return new Promise(function(resolve, reject) {
      let useBulk = canUseBulk(initialValues);

      const cmd = useBulk
        ? new BatchBulk(resolve, reject, _options, opts, _sql, initialValues)
        : new BatchRewrite(
            resolve,
            reject,
            _options,
            opts,
            _sql,
            initialValues
          );
      if (opts.trace) Error.captureStackTrace(cmd);
      _addCommand(cmd);
    });
  };

  /**
   * Execute query returning a Readable Object that will emit columns/data/end/error events
   * to permit streaming big result-set
   *
   * @param sql     sql parameter Object can be used to supersede default option.
   *                Object must then have sql property.
   * @param values  object / array of placeholder values (not mandatory)
   * @returns {Readable}
   */
  this.queryStream = (sql, values) => {
    let _cmdOpt, _sql;
    if (typeof sql === 'object') {
      _cmdOpt = sql;
      _sql = _cmdOpt.sql;
    } else {
      _sql = sql;
    }

    const cmd = new Stream(_cmdOpt, opts, _sql, values, _socket);
    if (opts.trace) Error.captureStackTrace(cmd);
    _addCommand(cmd);
    return cmd.inStream;
  };

  /**
   * Send an empty MySQL packet to ensure connection is active, and reset @@wait_timeout
   *
   * @returns {Promise} promise
   */
  this.ping = () => {
    return new Promise(function(resolve, reject) {
      return _addCommand(new Ping(resolve, reject));
    });
  };

  /**
   * Send a reset command that will
   * - rollback any open transaction
   * - reset transaction isolation level
   * - reset session variables
   * - delete user variables
   * - remove temporary tables
   * - remove all PREPARE statement
   *
   * @returns {Promise} promise
   */
  this.reset = () => {
    if (
      (info.isMariaDB() && info.hasMinVersion(10, 2, 4)) ||
      (!info.isMariaDB() && info.hasMinVersion(5, 7, 3))
    ) {
      return new Promise(function(resolve, reject) {
        return _addCommand(new Reset(resolve, reject));
      });
    }
    return Promise.reject(
      new Error(
        'Reset command not permitted for server ' + this.info.serverVersion
      )
    );
  };

  /**
   * Indicates the state of the connection as the driver knows it
   * @returns {boolean}
   */
  this.isValid = () => {
    return _status === Status.CONNECTED;
  };

  /**
   * Terminate connection gracefully.
   *
   * @returns {Promise} promise
   */
  this.end = () => {
    _addCommand = _addCommandDisabled;
    clearTimeout(_timeout);

    if (
      _status !== Status.CLOSING &&
      _status !== Status.CLOSED &&
      _status !== Status.NOT_CONNECTED
    ) {
      _status = Status.CLOSING;
      return new Promise(function(resolve, reject) {
        const ended = () => {
          _status = Status.CLOSED;
          _socket.destroy();
          _socket.unref();
          _clear();
          _receiveQueue.clear();
          resolve();
        };
        const quitCmd = new Quit(ended, ended);
        _sendQueue.push(quitCmd);
        _receiveQueue.push(quitCmd);
        if (_sendQueue.length === 1) {
          process.nextTick(_nextSendCmd.bind(this));
        }
      });
    }
    return Promise.resolve();
  };

  /**
   * Alias for destroy.
   */
  this.close = function() {
    this.destroy();
  };

  /**
   * Force connection termination by closing the underlying socket and killing server process if any.
   */
  this.destroy = () => {
    _addCommand = _addCommandDisabled;
    clearTimeout(_timeout);
    if (_status !== Status.CLOSING && _status !== Status.CLOSED) {
      _status = Status.CLOSING;
      _sendQueue.clear();
      if (_receiveQueue.length > 0) {
        //socket is closed, but server may still be processing a huge select
        //only possibility is to kill process by another thread
        //TODO reuse a pool connection to avoid connection creation
        const self = this;
        const killCon = new Connection(opts);
        killCon
          .connect()
          .then(() => {
            //*************************************************
            //kill connection
            //*************************************************
            const killResHandler = () => {
              const destroyError = Errors.createError(
                'Connection destroyed, command was killed',
                true,
                info,
                '08S01',
                Errors.ER_CMD_NOT_EXECUTED_DESTROYED
              );
              socketErrorDispatchToQueries(destroyError);
              process.nextTick(() => {
                if (_socket) _socket.destroy();
              });
              _status = Status.CLOSED;
              killCon.end().catch(() => {});
            };

            killCon
              .query('KILL ' + info.threadId)
              .then(killResHandler)
              .catch(killResHandler);
          })
          .catch(err => {
            //*************************************************
            //failing to create a kill connection, end normally
            //*************************************************
            const ended = () => {
              let sock = _socket;
              _clear();
              _status = Status.CLOSED;
              setImmediate(resolve);
              sock.destroy();
              _receiveQueue.clear();
            };
            const quitCmd = new Quit(ended, ended);
            _sendQueue.push(quitCmd);
            _receiveQueue.push(quitCmd);
            if (_sendQueue.length === 1) {
              process.nextTick(_nextSendCmd.bind(self));
            }
          });
      } else {
        _status = Status.CLOSED;
        _socket.destroy();
      }
    }
    _clear();
  };

  this.pause = () => {
    _socket.pause();
  };

  this.resume = () => {
    _socket.resume();
  };

  this.escape = value => {
    throw Errors.createError(
      '"Connection.escape intentionally not implemented. please use Connection.query(sql, values), it will be more secure and faster',
      false,
      info,
      '0A000',
      Errors.ER_NOT_IMPLEMENTED_ESCAPE
    );
  };

  this.escapeId = value => {
    throw Errors.createError(
      '"Connection.escapeId intentionally not implemented. please use Connection.query(sql, values), it will be more secure and faster',
      false,
      info,
      '0A000',
      Errors.ER_NOT_IMPLEMENTED_ESCAPEID
    );
  };

  this.format = (sql, values) => {
    throw Errors.createError(
      '"Connection.format intentionally not implemented. please use Connection.query(sql, values), it will be more secure and faster',
      false,
      info,
      '0A000',
      Errors.ER_NOT_IMPLEMENTED_FORMAT
    );
  };

  //*****************************************************************
  // additional public methods
  //*****************************************************************

  /**
   * return current connected server version information.
   *
   * @returns {*}
   */
  this.serverVersion = () => {
    if (!info.serverVersion)
      throw new Error(
        'cannot know if server information until connection is established'
      );
    return info.serverVersion.raw;
  };

  /**
   * Change option "debug" during connection.
   * @param val   debug value
   */
  this.debug = val => {
    opts.debug = val;
  };

  this.debugCompress = val => {
    opts.debugCompress = val;
  };

  //*****************************************************************
  // internal public testing methods
  //*****************************************************************

  function TestMethods() {}
  TestMethods.prototype.getCollation = () => {
    return opts.collation;
  };

  TestMethods.prototype.getSocket = () => {
    return _socket;
  };

  this.__tests = new TestMethods();

  //*****************************************************************
  // internal methods
  //*****************************************************************

  this._status = () => {
    return _status;
  };

  /**
   * Execute query using text protocol with callback emit columns/data/end/error
   * events to permit streaming big result-set
   *
   * @param sql     sql parameter Object can be used to supersede default option.
   *                Object must then have sql property.
   * @param values  object / array of placeholder values (not mandatory)
   * @param cb      callback
   * @returns {Query} query
   */
  this._queryCallback = (sql, values, cb) => {
    let _cmdOpts, _sql, _values;
    let _cb = cb;
    if (typeof sql === 'object') {
      _cmdOpts = sql;
      _sql = _cmdOpts.sql;
    } else {
      _sql = sql;
    }

    if (typeof values === 'function') {
      _cb = values;
    } else if (values !== undefined) {
      _values = values;
    }

    let cmd;
    if (_cb) {
      const resolve = rows => {
        const meta = rows.meta;
        delete rows.meta;
        _cb(null, rows, meta);
      };
      cmd = new Query(resolve, _cb, _cmdOpts, opts, _sql, _values);
    } else {
      cmd = new Query(() => {}, () => {}, _cmdOpts, opts, _sql, _values);
    }
    cmd.handleNewRows = row => {
      cmd._rows[cmd._responseIndex].push(row);
      cmd.emit('data', row);
    };

    if (opts.trace) Error.captureStackTrace(cmd);
    _addCommand(cmd);
    return cmd;
  };

  /**
   * Execute a batch using text protocol with callback emit columns/data/end/error
   * events to permit streaming big result-set
   *
   * @param sql     sql parameter Object can be used to supersede default option.
   *                Object must then have sql property.
   * @param values  object / array of placeholder values (not mandatory)
   * @param cb      callback
   * @returns {Query} query
   */
  this._batchCallback = (sql, values, cb) => {
    let _cmdOpts, _sql, _values;
    let _cb = cb;
    if (typeof sql === 'object') {
      _cmdOpts = sql;
      _sql = _cmdOpts.sql;
    } else {
      _sql = sql;
    }

    if (typeof values === 'function') {
      _cb = values;
    } else if (values !== undefined) {
      _values = values;
    }

    let cmd;

    let useBulk = canUseBulk(_values);

    const fct = useBulk ? BatchBulk : BatchRewrite;

    if (_cb) {
      const resolve = rows => {
        const meta = rows.meta;
        delete rows.meta;
        _cb(null, rows, meta);
      };
      cmd = new fct(resolve, _cb, _cmdOpts, opts, _sql, _values);
    } else {
      cmd = new fct(() => {}, () => {}, _cmdOpts, opts, _sql, _values);
    }
    cmd.handleNewRows = row => {
      cmd._rows[cmd._responseIndex].push(row);
      cmd.emit('data', row);
    };

    if (opts.trace) Error.captureStackTrace(cmd);
    _addCommand(cmd);
    return cmd;
  };

  /**
   * Use Batch rewrite or MariaDB bulk protocol.
   *
   * @param _values current batch values
   * @return {boolean} indicating if must use rewrite or bulk
   */
  const canUseBulk = _values => {
    let useBulk = info.isMariaDB() && info.hasMinVersion(10, 2, 7) && opts.bulk;
    if (useBulk) {
      //ensure that there is no stream object
      if (_values !== undefined) {
        const values = Array.isArray(_values) ? _values : [_values];
        const paramLen = Array.isArray(values[0]) ? values[0].length : 1;
        if (!opts.namedPlaceholders) {
          for (let r = 0; r < values.length; r++) {
            let row = values[r];
            if (!Array.isArray(row)) row = [row];
            if (paramLen !== row.length) {
              useBulk = false;
              break;
            }
            for (let j = 0; j < paramLen; j++) {
              const val = row[j];
              if (
                val !== null &&
                typeof val === 'object' &&
                typeof val.pipe === 'function' &&
                typeof val.read === 'function'
              ) {
                useBulk = false;
                break;
              }
            }
            if (!useBulk) break;
          }
        } else {
          for (let r = 0; r < values.length; r++) {
            let row = values[r];
            const keys = Object.keys(row);
            for (let j = 0; j < keys.length; j++) {
              const val = row[keys[j]];
              if (
                val !== null &&
                typeof val === 'object' &&
                typeof val.pipe === 'function' &&
                typeof val.read === 'function'
              ) {
                useBulk = false;
                break;
              }
            }
            if (!useBulk) break;
          }
        }
      }
    }
    return useBulk;
  };

  /**
   * Add handshake command to queue.
   *
   * @private
   */
  const _registerHandshakeCmd = (resolve, rejected) => {
    const _authFail = _authFailHandler.bind(this, rejected);
    const _authSucceed = _authSucceedHandler.bind(this, resolve, _authFail);

    const handshake = new Handshake(
      _authSucceed,
      _authFail,
      _createSecureContext.bind(this, _authFail),
      _addCommand.bind(this),
      _getSocket
    );
    Error.captureStackTrace(handshake);

    handshake.once('end', () => {
      process.nextTick(_nextSendCmd);
    });

    _receiveQueue.push(handshake);
    _initSocket(_authFail);
  };

  const _executeSessionVariableQuery = () => {
    if (opts.sessionVariables) {
      const values = [];
      let sessionQuery = 'set ';
      let keys = Object.keys(opts.sessionVariables);
      if (keys.length > 0) {
        return new Promise(function(resolve, reject) {
          for (let k = 0; k < keys.length; ++k) {
            sessionQuery +=
              (k !== 0 ? ',' : '') +
              '@@' +
              keys[k].replace(/[^a-z0-9_]/gi, '') +
              '=?';
            values.push(opts.sessionVariables[keys[k]]);
          }
          const errorHandling = initialErr => {
            reject(
              Errors.createError(
                'Error setting session variable (value ' +
                  JSON.stringify(opts.sessionVariables) +
                  '). Error: ' +
                  initialErr.message,
                true,
                info,
                '08S01',
                Errors.ER_SETTING_SESSION_ERROR,
                null
              )
            );
          };
          const cmd = new Query(
            resolve,
            errorHandling,
            null,
            opts,
            sessionQuery,
            values
          );
          if (opts.trace) Error.captureStackTrace(cmd);
          _addCommand(cmd);
        });
      }
    }
    return Promise.resolve();
  };

  /**
   * Asking server timezone if not set
   * @returns {Promise<void>|Promise<T | never>}
   * @private
   */
  const _checkServerTimezone = () => {
    if (opts.timezone === 'local' || opts.tz) {
      return Promise.resolve();
    }
    return this.query('SELECT @@system_time_zone stz, @@time_zone tz').then(
      res => {
        const serverTimezone = res[0].tz === 'SYSTEM' ? res[0].stz : res[0].tz;
        const serverZone = moment.tz.zone(serverTimezone);
        if (serverZone) {
          if (serverTimezone === moment.tz.guess()) {
            //db server and client use same timezone, avoid any conversion
            opts.tz = null;
          } else {
            opts.tz = moment.tz.setDefault(serverTimezone);
          }
        } else {
          return Promise.reject(
            Errors.createError(
              "Automatic timezone setting fails. Server timezone '" +
                serverTimezone +
                "' does't have a corresponding IANA timezone. Option timezone must be set according to server timezone",
              true,
              info,
              '08S01',
              Errors.ER_WRONG_AUTO_TIMEZONE
            )
          );
        }
        return Promise.resolve();
      }
    );
  };

  const _executeInitQuery = () => {
    if (opts.initSql) {
      const initialArr = Array.isArray(opts.initSql)
        ? opts.initSql
        : [opts.initSql];
      const initialPromises = [];
      initialArr.forEach(sql => {
        initialPromises.push(
          new Promise(function(resolve, reject) {
            const errorHandling = initialErr => {
              reject(
                Errors.createError(
                  'Error executing initial sql command: ' + initialErr.message,
                  true,
                  info,
                  '08S01',
                  Errors.ER_INITIAL_SQL_ERROR,
                  null
                )
              );
            };
            const cmd = new Query(
              resolve,
              errorHandling,
              null,
              opts,
              sql,
              null
            );
            if (opts.trace) Error.captureStackTrace(cmd);
            _addCommand(cmd);
          })
        );
      });
      return Promise.all(initialPromises);
    }
    return Promise.resolve();
  };

  const _getSocket = () => {
    return _socket;
  };

  /**
   * Initialize socket and associate events.
   * @private
   */
  const _initSocket = authFailHandler => {
    if (opts.socketPath) {
      _socket = Net.connect(opts.socketPath);
    } else {
      _socket = Net.connect(opts.port, opts.host);
    }

    if (opts.connectTimeout) {
      _timeout = setTimeout(
        _connectTimeoutReached,
        opts.connectTimeout,
        authFailHandler
      );
    }

    const _socketError = _socketErrorHandler.bind(this, authFailHandler);

    _socket.on('data', _in.onData.bind(_in));
    _socket.on('error', _socketError);
    _socket.on('end', _socketError);
    _socket.on(
      'connect',
      function() {
        clearTimeout(_timeout);
        if (_status === Status.CONNECTING) {
          _status = Status.AUTHENTICATING;
          _socketConnected = true;
          _socket.setTimeout(
            opts.socketTimeout,
            _socketTimeoutReached.bind(this, authFailHandler)
          );
          _socket.setNoDelay(true);
        }
      }.bind(this)
    );

    _socket.writeBuf = _socket.write;
    _socket.flush = () => {};
    _out.setStream(_socket);
  };

  /**
   * Authentication success result handler.
   *
   * @private
   */
  const _authSucceedHandler = (resolve, rejected) => {
    //enable packet compression according to option
    if (opts.logPackets) info.enableLogPacket();
    if (opts.compress) {
      if (info.serverCapabilities & Capabilities.COMPRESS) {
        _out.setStream(new CompressionOutputStream(_socket, opts, info));
        _in = new CompressionInputStream(_in, _receiveQueue, opts, info);
        _socket.removeAllListeners('data');
        _socket.on('data', _in.onData.bind(_in));
      } else {
        console.error(
          "connection is configured to use packet compression, but the server doesn't have this capability"
        );
      }
    }

    if (opts.pipelining) {
      _addCommand = _addCommandEnablePipeline;
      const commands = _receiveQueue.toArray();
      commands.forEach(cmd => {
        const listeners = cmd.listeners('end');
        cmd.removeListener('end', listeners[0]);
        cmd.once('send_end', () => {
          setImmediate(_nextSendCmd);
        });
      });
    }

    const errorInitialQueries = err => {
      if (!err.fatal) this.end().catch(err => {});
      process.nextTick(rejected, err);
    };

    _executeSessionVariableQuery()
      .then(() => {
        return _checkServerTimezone();
      })
      .then(() => {
        return _executeInitQuery();
      })
      .then(() => {
        _status = Status.CONNECTED;
        process.nextTick(resolve, this);
      })
      .catch(errorInitialQueries);
  };

  /**
   * Authentication failed result handler.
   *
   * @private
   */
  const _authFailHandler = (reject, err) => {
    process.nextTick(reject, err);
    //remove handshake command
    _receiveQueue.shift();

    _fatalError(err, true);
  };

  /**
   * Create TLS socket and associate events.
   *
   * @param rejected  rejected function when error
   * @param callback  callback function when done
   * @private
   */
  const _createSecureContext = (rejected, callback) => {
    const _socketError = _socketErrorHandler.bind(this, rejected);
    const sslOption = Object.assign({}, opts.ssl, {
      servername: opts.host,
      socket: _socket
    });

    try {
      const secureSocket = tls.connect(sslOption, callback);

      secureSocket.on('data', _in.onData.bind(_in));
      secureSocket.on('error', _socketError);
      secureSocket.on('end', _socketError);
      secureSocket.writeBuf = secureSocket.write;
      secureSocket.flush = () => {};

      _socket.removeAllListeners('data');
      _socket = secureSocket;

      _out.setStream(secureSocket);
    } catch (err) {
      _socketError(err);
    }
  };

  /**
   * Handle packet when no packet is expected.
   * (there can be an ERROR packet send by server/proxy to inform that connection is ending).
   *
   * @param packet  packet
   * @private
   */
  const _unexpectedPacket = function(packet) {
    if (packet && packet.peek() === 0xff) {
      //can receive unexpected error packet from server/proxy
      //to inform that connection is closed (usually by timeout)
      let err = packet.readError(info);
      if (
        err.fatal &&
        (_status !== Status.CLOSING && _status !== Status.CLOSED)
      ) {
        this.emit('error', err);
        this.end();
      }
    } else if (_status !== Status.CLOSING && _status !== Status.CLOSED) {
      this.emit(
        'error',
        Errors.createError(
          'receiving packet from server without active commands\n' +
            'conn:' +
            (info.threadId ? info.threadId : -1) +
            '(' +
            packet.pos +
            ',' +
            packet.end +
            ')\n' +
            Utils.log(opts, packet.buf, packet.pos, packet.end),
          true,
          info,
          '08S01',
          Errors.ER_UNEXPECTED_PACKET
        )
      );
      this.destroy();
    }
  };

  /**
   * Change transaction state.
   *
   * @param sql sql
   * @returns {Promise} promise
   * @private
   */
  const _changeTransaction = sql => {
    //if command in progress, driver cannot rely on status and must execute query
    if (_status === Status.CLOSING || _status === Status.CLOSED) {
      return Promise.reject(
        Errors.createError(
          'Cannot execute new commands: connection closed\nsql: ' + sql,
          true,
          info,
          '08S01',
          Errors.ER_CMD_CONNECTION_CLOSED
        )
      );
    }

    //Command in progress => must execute query
    //or if no command in progress, can rely on status to know if query is needed
    if (
      _receiveQueue.peekFront() ||
      info.status & ServerStatus.STATUS_IN_TRANS
    ) {
      return new Promise(function(resolve, reject) {
        const cmd = new Query(resolve, reject, null, opts, sql, null);
        if (opts.trace) Error.captureStackTrace(cmd);
        _addCommand(cmd);
      });
    }
    return Promise.resolve();
  };

  /**
   * Handle connection timeout.
   *
   * @private
   */
  const _connectTimeoutReached = function(authFailHandler) {
    _timeout = null;
    const handshake = _receiveQueue.peekFront();
    authFailHandler(
      Errors.createError(
        'Connection timeout',
        true,
        info,
        '08S01',
        Errors.ER_CONNECTION_TIMEOUT,
        handshake ? handshake.stack : null
      )
    );
  };

  /**
   * Handle socket timeout.
   *
   * @private
   */
  const _socketTimeoutReached = function() {
    const err = Errors.createError(
      'socket timeout',
      true,
      info,
      '08S01',
      Errors.ER_SOCKET_TIMEOUT
    );
    const packetMsgs = info.getLastPackets();
    if (packetMsgs !== '') {
      err.message = err.message + '\nlast received packets:\n' + packetMsgs;
    }
    _fatalError(err, true);
  };

  /**
   * Add command to command sending and receiving queue.
   *
   * @param cmd         command
   * @returns {*}       current command
   * @private
   */
  const _addCommandEnable = cmd => {
    cmd.once('end', () => {
      setImmediate(_nextSendCmd);
    });

    //send immediately only if no current active receiver
    if (_sendQueue.isEmpty() && _status === Status.CONNECTED) {
      if (_receiveQueue.peekFront()) {
        _receiveQueue.push(cmd);
        _sendQueue.push(cmd);
        return cmd;
      }

      _receiveQueue.push(cmd);
      cmd.start(_out, opts, info);
    } else {
      _receiveQueue.push(cmd);
      _sendQueue.push(cmd);
    }
    return cmd;
  };

  /**
   * Add command to command sending and receiving queue using pipelining
   *
   * @param cmd         command
   * @returns {*}       current command
   * @private
   */
  const _addCommandEnablePipeline = cmd => {
    cmd.once('send_end', () => {
      setImmediate(_nextSendCmd);
    });

    _receiveQueue.push(cmd);
    if (_sendQueue.isEmpty()) {
      cmd.start(_out, opts, info);
      if (cmd.sending) {
        _sendQueue.push(cmd);
        cmd.prependOnceListener('send_end', () => {
          _sendQueue.shift();
        });
      }
    } else {
      _sendQueue.push(cmd);
    }
    return cmd;
  };

  /**
   * Replacing command when connection is closing or closed to send a proper error message.
   *
   * @param cmd         command
   * @private
   */
  const _addCommandDisabled = cmd => {
    cmd.throwNewError(
      'Cannot execute new commands: connection closed\n' + cmd.displaySql(),
      true,
      info,
      '08S01',
      Errors.ER_CMD_CONNECTION_CLOSED
    );
  };

  /**
   * Handle socket error.
   *
   * @param authFailHandler   authentication handler
   * @param err               socket error
   * @private
   */
  const _socketErrorHandler = function(authFailHandler, err) {
    if (_status === Status.CLOSING || _status === Status.CLOSED) return;

    _socket.writeBuf = () => {};
    _socket.flush = () => {};

    //socket has been ended without error
    if (!err) {
      err = Errors.createError(
        'socket has unexpectedly been closed',
        true,
        info,
        '08S01',
        Errors.ER_SOCKET_UNEXPECTED_CLOSE
      );
    } else {
      err.fatal = true;
      this.sqlState = 'HY000';
    }

    const packetMsgs = info.getLastPackets();
    if (packetMsgs !== '') {
      err.message += '\nlast received packets:\n' + packetMsgs;
    }

    switch (_status) {
      case Status.CONNECTING:
      case Status.AUTHENTICATING:
        const currentCmd = _receiveQueue.peekFront();
        if (currentCmd && currentCmd.stack && err) {
          err.stack +=
            '\n From event:\n' +
            currentCmd.stack.substring(currentCmd.stack.indexOf('\n') + 1);
        }
        authFailHandler(err);
        break;

      default:
        _fatalError(err, false);
    }
  };

  /**
   * Fatal unexpected error : closing connection, and throw exception.
   *
   * @param self    current connection
   * @private
   */
  const _fatalErrorHandler = function(self) {
    return function(err, avoidThrowError) {
      if (_status === Status.CLOSING || _status === Status.CLOSED) {
        socketErrorDispatchToQueries(err);
        return;
      }
      const mustThrowError = _status !== Status.CONNECTING;
      _status = Status.CLOSING;

      //prevent executing new commands
      _addCommand = _addCommandDisabled;

      if (_socket) {
        _socket.removeAllListeners('error');
        _socket.removeAllListeners('timeout');
        _socket.removeAllListeners('close');
        _socket.removeAllListeners('data');
        if (!_socket.destroyed) _socket.destroy();
        _socket = undefined;
      }
      _status = Status.CLOSED;

      const errorThrownByCmd = socketErrorDispatchToQueries(err);
      if (mustThrowError) {
        if (self.listenerCount('error') > 0) {
          self.emit('error', err);
          self.emit('end');
          _clear();
        } else {
          self.emit('end');
          _clear();
          //error will be thrown if no error listener and no command did throw the exception
          if (!avoidThrowError && !errorThrownByCmd) throw err;
        }
      } else {
        _clear();
      }
    };
  };

  /**
   * Dispatch fatal error to current running queries.
   *
   * @param err        the fatal error
   * @return {boolean} return if error has been relayed to queries
   */
  const socketErrorDispatchToQueries = err => {
    let receiveCmd;
    let errorThrownByCmd = false;
    while ((receiveCmd = _receiveQueue.shift())) {
      if (receiveCmd && receiveCmd.onPacketReceive) {
        errorThrownByCmd = true;
        setImmediate(receiveCmd.throwError.bind(receiveCmd), err, info);
      }
    }
    return errorThrownByCmd;
  };

  /**
   * Will send next command in queue if any.
   *
   * @private
   */
  const _nextSendCmd = () => {
    let sendCmd;
    if ((sendCmd = _sendQueue.shift())) {
      if (sendCmd.sending) {
        _sendQueue.unshift(sendCmd);
      } else {
        sendCmd.start(_out, opts, info);
        if (sendCmd.sending) {
          sendCmd.prependOnceListener('send_end', () => {
            _sendQueue.shift();
          });
          _sendQueue.unshift(sendCmd);
        }
      }
    }
  };

  /**
   * Clearing connection variables when ending.
   *
   * @private
   */
  const _clear = () => {
    _sendQueue.clear();
    _out = undefined;
    _socket = undefined;
  };

  //*****************************************************************
  // internal variables
  //*****************************************************************

  EventEmitter.call(this);
  const opts = Object.assign({}, options);
  const info = new ConnectionInformation();
  const _sendQueue = new Queue();
  const _receiveQueue = new Queue();
  let _status = Status.NOT_CONNECTED;
  let _socketConnected = false;
  let _socket = null;
  let _timeout = null;
  let _addCommand = _addCommandEnable;
  const _fatalError = _fatalErrorHandler(this);
  let _out = new PacketOutputStream(opts, info);
  let _in = new PacketInputStream(
    _unexpectedPacket.bind(this),
    _receiveQueue,
    _out,
    opts,
    info
  );

  //add alias threadId for mysql/mysql2 compatibility
  Object.defineProperty(this, 'threadId', {
    get() {
      return info ? info.threadId : undefined;
    }
  });
  Object.defineProperty(this, 'info', {
    get() {
      return info;
    }
  });
}
482517
function validateConnection(connection) {
  const coreFields = ['_id', 'name', 'driver', 'createdDate', 'modifiedDate'];
  if (!connection.name) {
    throw new Error('connection.name required');
  }
  if (!connection.driver) {
    throw new Error('connection.driver required');
  }
  const driver = drivers[connection.driver];
  if (!driver) {
    throw new Error(`driver implementation ${connection.driver} not found`);
  }
  const validFields = driver.fields.map(field => field.key).concat(coreFields);
  const cleanedConnection = validFields.reduce(
    (cleanedConnection, fieldKey) => {
      if (connection.hasOwnProperty(fieldKey)) {
        let value = connection[fieldKey];
        const fieldDefinition =
          drivers[connection.driver].fieldsByKey[fieldKey];

        // field definition may not exist since
        // this could be a core field like _id, name
        if (fieldDefinition) {
          if (fieldDefinition.formType === 'CHECKBOX') {
            value = utils.ensureBoolean(value);
          }
        }

        cleanedConnection[fieldKey] = value;
      }
      return cleanedConnection;
    },
    {}
  );

  return cleanedConnection;
}
1105891
protected final static function attachHooks( $connector, $hooksList )
    {
        if ( is_array( $hooksList ) && count( $hooksList ) )
        {
            foreach( $hooksList as $hook )
            {
                self::signals()->connect( $connector, $hook );
            }
        }
    }
859095
function redditConnector( parentDirPath ){

	//Call constructor from super class
	connectorExt.call(this, 'reddit', 'Reddit AMA', {
		extraRequiredFields: 'amaURL',
		useOAuth: false,
		useCustomTables: true
	});
	
	this.getTablePrefix = function(){
		return 'reddit';
	};
	
	this.getCloudantDbName = function(pipe, table){
		return pipe.name + '_' + table.name;
	};

	var watsonToneAnalyzerService = vcapServices.getService( configManager.get('WATSON_TONE_ANALYZER') || 'tone analyzer' );
	
	this.fetchRecords = function( table, pushRecordFn, done, pipeRunStep, pipeRunStats, logger, pipe, pipeRunner ){

		if ( !watsonToneAnalyzerService ){
			var msg = 'Unable to find Watson Tone Analyzer Service';
			logger.error( msg );
			return done( msg );
		}

		if (( !pipe.amaURL ) || (pipe.amaURL.length < 1)){
			return done('Missing or empty field amaURL');
		}

		var amaJsonURL = '';
		// append json suffix to amaURL
		// input: https://www.reddit.com/r/IAmA/comments/3ilzey/were_a_bunch_of_developers_from_ibm_ask_us.json or
		//        https://www.reddit.com/r/IAmA/comments/3ilzey/were_a_bunch_of_developers_from_ibm_ask_us or
		//        https://www.reddit.com/r/IAmA/comments/3ilzey/were_a_bunch_of_developers_from_ibm_ask_us/ 
		// output https://www.reddit.com/r/IAmA/comments/3ilzey/were_a_bunch_of_developers_from_ibm_ask_us.json

        if(pipe.amaURL.match(/\.json$/)) {
        	amaJsonURL = pipe.amaURL;
        }
        else {
			if(pipe.amaURL.charAt(pipe.amaURL.length-1) == '/') {
				amaJsonURL = pipe.amaURL.replace(/\/+$/,'.json');
			}
			else {
				amaJsonURL = pipe.amaURL + '.json';
			}
		}

		// fetch up to 500 comments by default, sorted by popularity
		amaJsonURL = amaJsonURL + '?limit=500&showmore=true&sort=top';

		if(table.id === 'top_comments_only') {
			// limit depth to fetch more records at the highest level
			amaJsonURL = amaJsonURL + '&depth=1';  
		}

		logger.info('Connecting to AMA with url ' + amaJsonURL);

		request.get(amaJsonURL, {'json': true},
			function( err, response, body ){
				if ( err ){
					logger.error('Unable to fetch AMA: ' + err );
					return done( err );
				}

				// Create an async queue to process the record with Watson Tone Analyzer
				var q = async.queue( function( post, callback ){

					var msg = '';

					// Call Tone analyzer to get sentiments from the post
					// https://www.ibm.com/smarterplanet/us/en/ibmwatson/developercloud/doc/tone-analyzer/output.shtml
					request.post( watsonToneAnalyzerService.credentials.url + '/v3/tone?version=2016-02-11',{
						'auth': {
						  'user': watsonToneAnalyzerService.credentials.username,
						  'pass': watsonToneAnalyzerService.credentials.password,
						  'sendImmediately': true
						},
						'json': {
							'text': post.selftext || post.body
						}
					}, function( err, response, body ){

						if ( err ){
							msg = 'Error querying Watson Tone Analyzer service: ' + err;
							logger.error(msg);
							return callback( msg );
						}

						// logger.debug('WTA response: ' + util.inspect(response));	
						// logger.debug('WTA body: ' + util.inspect(body));						

						if(response.statusCode >= 300) {
							msg = 'Call to Watson Tone Analyzer URL ' + response.request.uri.href + ' returned status ' + response.statusCode + ' (' + response.body.error + ')';
							logger.error(msg);
							logger.info('FFDC - post: ' + util.inspect(post,5));
							return callback(msg);
						}
						
						var sentiments = {
											author: post.author,
											created: post.created_utc,
											edited: post.edited,
											id: post.id,
											num_comments: post.num_comments,
											title: post.title || '',
											text: post.selftext || post.body
                                         };

                        if(body.document_tone) {

                        	/*

                        	Tone Analyzer API v3 output (http://www.ibm.com/smarterplanet/us/en/ibmwatson/developercloud/tone-analyzer/api/v3/)
							
							{ 
							    document_tone: {
	
												tone_categories: [ 
																  	{ 
																  	  tones: [ [Object], [Object], [Object], [Object], [Object] ], 
																  	  category_id: 'emotion_tone', 
																      category_name: 'Emotion Tone' 
																    }, 
																    { 
																      tones: [ [Object], [Object], [Object] ], 
																      category_id: 'writing_tone', 
																      category_name: 'Writing Tone' 
																    }, 
																    { 
																      tones: [ [Object], [Object], [Object], [Object], [Object] ], 
																      category_id: 'socialevel_tone', 
																      category_name: 'Social Tone' 
																    } 
															      ]
										        },
							    sentences_tone: ...

							}			     

                        	*/

							// logger.debug('Document tone categories: ' + util.inspect(body.document_tone.tone_categories));	

							var generate_flat_document_structure = (pipe.outputFormat === 'JSON_flat');

							if(generate_flat_document_structure) {
								_.forEach(body.document_tone.tone_categories, function(tone_category) {
									_.forEach(tone_category.tones, function(tone){
										// Sample tone definition: { score: 0.10796, tone_id: 'anger', tone_name: 'Anger' }
										sentiments[tone.tone_name.replace(/ /g,'_')] = (parseFloat(tone.score * 100)).toFixed(2);
									});
								});

							}
							else {
								_.forEach(body.document_tone.tone_categories, function(tone_category) {
									sentiments[tone_category.category_id] = { 
										                                      category_name : tone_category.category_name,
										                                      tones: tone_category.tones 
										                                    };
									// add normalized score (between 0 and 100, two decimal places)	                                    
									_.forEach(sentiments[tone_category.category_id].tones, function(tone){
										// Sample tone definition: { score: 0.10796, tone_id: 'anger', tone_name: 'Anger' }
										tone.normalized_score = (parseFloat(tone.score * 100)).toFixed(2);
									});	                                    
								});	

							}

                        	// save in Cloudant
                            pushRecordFn(sentiments); 	

                        }   
                        else {
                        	// No tone analyzer results were returned
                        	// Don't store post in Cloudant
                        	logger.warn('No Tone Analyzer results were returned for ' + sentiments.title + '.'); 
                        }          

   						return callback();
		        	
                    }); // post
				}, concurrency ); // async

				var amaStats = {
									processed_record_count : 0,		      // total number of records sent to Tone Analyzer
									missing_record_count : 0,             // total number of records for which text needs to be manually loaded
									max_processed_level : 1,        	  // identifies how many levels of comments were processed (1 is minimum)
									level_stats : {},				      // statistics for each level within the thread
									parser_warnings : 0                   // if non-zero, potential parsing issues were detected
								};

				/*
						comment1              (level 1)
						  comment1.1          (level 2)
						  comment1.2          (level 2)
						    comment1.2.1      (level 3)
						comment2              (level 1)
						  comment2.1          (level 2)
						    comment2.1.1      (level 3)
						    comment2.1.1.1    (level 4)
						 comment3             (level 1)

				*/	

				// If the user chose option top_comments, things (comments at level 2+) are skipped			
				if(table.id === 'top_comments_only') {
					// process only comments at level 1
					logger.info('Analyzing only one level.');		
					amaStats.max_depth_to_inspect = 1;
				}				

				/*
				 * Local function. Collects comment information at the current level and invokes itself
				 *  to collect information at lower levels.
				 * @param things AMA records 
				 * @param level - current level
				 */
				var processSubtree = function(things, level) {
						// stop if there is nothing to process or if the maximum processing depth has been reached
						if(((! things) && (level > 1)) || ((amaStats.max_depth_to_inspect) && (level > amaStats.max_depth_to_inspect))){
							return;
						}

						// update statistics	
						if(amaStats.max_processed_level < level) {
							amaStats.max_processed_level = level;	
						}

						if(! amaStats.level_stats['level_' + level]) {
							amaStats.level_stats['level_' + level] = {
																		processed : 0,
																		warnings : 0,
																		missing	: 0
																	};
						}
						
						// process records 
						if(things.hasOwnProperty('kind') && (things.kind === 'Listing')) {
								if((things.hasOwnProperty('data')) && (things.data.hasOwnProperty('children'))) {
									_.forEach(things.data.children, function (thing) {
											// for a definition of thing refer to the Reddit API doc
											if(thing.hasOwnProperty('data')) {
												if((thing.data.hasOwnProperty('selftext')) || (thing.data.hasOwnProperty('body'))) {
													// the record includes text that can be analyzed; submit comment for Tone Analyzer processing
													q.push( thing.data );	
													amaStats.processed_record_count++;
													amaStats.level_stats['level_' + level].processed++;
												}
												else {
													// the record does not include text that can be analyzed; skip

													if(thing.kind === 'more'){
														// type "more" indicates that there are things (e.g. comments) in this subtree that were not loaded but could be 
														// retrieved using https://www.reddit.com/dev/api#GET_api_morechildren
												    	// TODO additional processing is required
												    	// calculate total (across all levels)
												    	if(thing.data.count) {
															amaStats.missing_record_count = amaStats.missing_record_count + thing.data.count; 
												    		// calculate total for current level
												    		amaStats.level_stats['level_' + level].missing = amaStats.level_stats['level_' + level].missing + thing.data.count;														 
												    	}
													}
													else {
														  // unknown record type 	
														  amaStats.parser_warnings++;
														  amaStats.level_stats['level_' + level].warnings++;
														  logger.warn('Potential parser error detected for record type ' + thing.kind + ' at level ' + level + ': ' + util.inspect(thing.data,3));
													}	
												}

												// process sub-tree things (which are typically comments/replies)
												// before processung other things at the current level
												processSubtree(thing.data.replies, level + 1);
											}	
									});
								} // ! data.children
								else {
									// potential parser logic error
									amaStats.parser_warnings++;
									amaStats.level_stats['level_' + level].warnings++;
									logger.warn('Potential parser error detected at level ' + level + ': ' + util.inspect(things,3));
								}
						}
						else {
								// potential parser logic error
								amaStats.parser_warnings++;
								amaStats.level_stats['level_' + level].warnings++;
								logger.warn('Potential parser error detected at level ' + level + ': ' + util.inspect(things,3));
						}
				};

				// inspect the AMA base tree
				logger.info('Processing AMA base tree for AMA URL ' + amaJsonURL);

				_.forEach(body, function (listing) {
					processSubtree(listing, 1);							
				});
				
				
				// local helper function; displays processing statistics`
				var logStats = function () {
					logger.info('Records sent to Tone Analyzer: ' + amaStats.processed_record_count);
					// these records could be fetched using other API calls
					logger.info('Records missing (no comment text available): ' + amaStats.missing_record_count);
					logger.info('Maximum processed comment thread depth: ' + amaStats.max_processed_level);
					if(amaStats.parser_warnings > 0) {
						// Potential issue with parser logic. If a non-zero value is reported, the base tree should be reviewed.
						logger.warn('Potential base tree parsing issues: ' + amaStats.parser_warnings);	
					}
					
					logger.debug('AMA thread statistics: ' + util.inspect(amaStats.level_stats,3));

				};

				// Drain function is invoked after all posts have been processed
				q.drain = function(){
					// pipe processing complete
					logStats();
					return done();
				};

				// if the queue is empty asynchronous processing has already completed (or there was nothing to process)
				if(q.idle()) {
					logStats();
					return done();
				}
			}
		);
	};
}
1057366
function createConnect() {
  var _ref = arguments.length > 0 && arguments[0] !== undefined ? arguments[0] : {},
      _ref$connectHOC = _ref.connectHOC,
      connectHOC = _ref$connectHOC === undefined ? __WEBPACK_IMPORTED_MODULE_0__components_connectAdvanced__["a" /* default */] : _ref$connectHOC,
      _ref$mapStateToPropsF = _ref.mapStateToPropsFactories,
      mapStateToPropsFactories = _ref$mapStateToPropsF === undefined ? __WEBPACK_IMPORTED_MODULE_3__mapStateToProps__["a" /* default */] : _ref$mapStateToPropsF,
      _ref$mapDispatchToPro = _ref.mapDispatchToPropsFactories,
      mapDispatchToPropsFactories = _ref$mapDispatchToPro === undefined ? __WEBPACK_IMPORTED_MODULE_2__mapDispatchToProps__["a" /* default */] : _ref$mapDispatchToPro,
      _ref$mergePropsFactor = _ref.mergePropsFactories,
      mergePropsFactories = _ref$mergePropsFactor === undefined ? __WEBPACK_IMPORTED_MODULE_4__mergeProps__["a" /* default */] : _ref$mergePropsFactor,
      _ref$selectorFactory = _ref.selectorFactory,
      selectorFactory = _ref$selectorFactory === undefined ? __WEBPACK_IMPORTED_MODULE_5__selectorFactory__["a" /* default */] : _ref$selectorFactory;

  return function connect(mapStateToProps, mapDispatchToProps, mergeProps) {
    var _ref2 = arguments.length > 3 && arguments[3] !== undefined ? arguments[3] : {},
        _ref2$pure = _ref2.pure,
        pure = _ref2$pure === undefined ? true : _ref2$pure,
        _ref2$areStatesEqual = _ref2.areStatesEqual,
        areStatesEqual = _ref2$areStatesEqual === undefined ? strictEqual : _ref2$areStatesEqual,
        _ref2$areOwnPropsEqua = _ref2.areOwnPropsEqual,
        areOwnPropsEqual = _ref2$areOwnPropsEqua === undefined ? __WEBPACK_IMPORTED_MODULE_1__utils_shallowEqual__["a" /* default */] : _ref2$areOwnPropsEqua,
        _ref2$areStatePropsEq = _ref2.areStatePropsEqual,
        areStatePropsEqual = _ref2$areStatePropsEq === undefined ? __WEBPACK_IMPORTED_MODULE_1__utils_shallowEqual__["a" /* default */] : _ref2$areStatePropsEq,
        _ref2$areMergedPropsE = _ref2.areMergedPropsEqual,
        areMergedPropsEqual = _ref2$areMergedPropsE === undefined ? __WEBPACK_IMPORTED_MODULE_1__utils_shallowEqual__["a" /* default */] : _ref2$areMergedPropsE,
        extraOptions = _objectWithoutProperties(_ref2, ['pure', 'areStatesEqual', 'areOwnPropsEqual', 'areStatePropsEqual', 'areMergedPropsEqual']);

    var initMapStateToProps = match(mapStateToProps, mapStateToPropsFactories, 'mapStateToProps');
    var initMapDispatchToProps = match(mapDispatchToProps, mapDispatchToPropsFactories, 'mapDispatchToProps');
    var initMergeProps = match(mergeProps, mergePropsFactories, 'mergeProps');

    return connectHOC(selectorFactory, _extends({
      // used in error messages
      methodName: 'connect',

      // used to compute Connect's displayName from the wrapped component's displayName.
      getDisplayName: function getDisplayName(name) {
        return 'Connect(' + name + ')';
      },

      // if mapStateToProps is falsy, the Connect component doesn't subscribe to store state changes
      shouldHandleStateChanges: Boolean(mapStateToProps),

      // passed through to selectorFactory
      initMapStateToProps: initMapStateToProps,
      initMapDispatchToProps: initMapDispatchToProps,
      initMergeProps: initMergeProps,
      pure: pure,
      areStatesEqual: areStatesEqual,
      areOwnPropsEqual: areOwnPropsEqual,
      areStatePropsEqual: areStatePropsEqual,
      areMergedPropsEqual: areMergedPropsEqual

    }, extraOptions));
  };
}
406661
function workFn() {
            if (currentSpec.$injector) {
                throw Error('Injector already created, can not register a module!');
            } else {
                var modules = currentSpec.$modules || (currentSpec.$modules = []);
                angular.forEach(moduleFns, function(module) {
                    if (angular.isObject(module) && !angular.isArray(module)) {
                        modules.push(function($provide) {
                            angular.forEach(module, function(value, key) {
                                $provide.value(key, value);
                            });
                        });
                    } else {
                        modules.push(module);
                    }
                });
            }
        }
209461
function createConnectionPromise(key, options) {
  
  debug('createConnectionPromise','start');

  //create promise that resolves to a connection
  var promise = new Promise((resolve,reject)=>{
    //create a new mssql connection
    let connection = new mssql.Connection(options,(err)=>handleConnectionCallback(err, resolve, reject, key, connection));
    connection.close = ()=>promise.then((conn)=>conn.close()).catch(()=>{ delete promises[key]; delete connections[key] });
    connection.query = (templateParts, ...params)=>promise.then((conn)=>query(conn,{},templateParts,params));
    connection.queryStream = (templateParts, ...params)=>promise.then((conn)=>query(conn,{stream:true},templateParts,params));
    connection.output = require('../parameters/output');
    connection.input = require('../parameters/input');

    debug('createConnectionPromise','success');
  });

  //add mssql types to promise - convenience access
  assign(promise, mssql.TYPES); //add mssql types to promise being returned
  promise.MAX = mssql.MAX;
  
  //back reference to cache
  promise._mssqlngKey = key;

  //extended methods
  promise.close = ()=>promise.then((conn)=>conn.close()).catch(()=>{ delete promises[key]; delete connections[key] });
  promise.query = (templateParts, ...params)=>promise.then((conn)=>query(conn,{},templateParts,params));
  promise.queryStream = (templateParts, ...params)=>promise.then((conn)=>query(conn,{stream:true},templateParts,params));
  promise.output = require('../parameters/output');
  promise.input = require('../parameters/input');

  debug('createConnectionPromise','success');
  return promise;
}
644210
function printScannedDataConnectionTypes(scannedData) {
        var keys = Object.keys(scannedData);
        keys.forEach(function(key) {
            var dev = scannedData[key];
            var cts = dev.connectionTypes;
            debugLFDS(key, cts);
        });
    }