function processAction(msg, cfg, snapshot = {}) {
  const originalSql = cfg.query || msg.body.query;
  const now = new Date().toISOString();
  // Last poll should come from Snapshot, if not it's beginning of time
  const lastPoll = snapshot.lastPoll || new Date(0).toISOString();
  this.logger.info('Last polling timestamp=%s', lastPoll);
  const sql = originalSql.split(LAST_POLL_PLACEHOLDER).join(lastPoll);
  this.logger.debug('Transformed query is ready');
  const that = this;
  return co(function* gen() {
    const request = new cosql.Request(connection);
    request.stream = true;

    request.on('recordset', () => {
      that.logger.trace('Have got recordset metadata');
    });

    request.on('row', (row) => {
      this.emit('data', eioUtils.newMessageWithBody(row));
    });

    request.on('error', (err) => {
      this.emit('error', err);
    });

    request.on('done', (affected) => {
      that.logger.info('Query execution completed, affected=%s', affected);
      this.emit('snapshot', {
        lastPoll: now,
      });
      this.emit('end');
    });

    // Run it
    yield request.query(sql);
  }.bind(this));
}
/**
 * This method will be called from elastic.io platform providing following data
 *
 * @param msg incoming message object that contains ``body`` with payload
 * @param cfg configuration that is account information and configuration field values
 */
This method will be called from elastic.io platform providing following data
@param msg incoming message object that contains ``body`` with payload
@param cfg configuration that is account information and configuration field values
This method will be called from elastic . io platform providing following data @param msg incoming message object that contains ` ` body ` ` with payload @param cfg configuration that is account information and configuration field values
function processAction(msg, cfg, snapshot = {}) {
  const originalSql = cfg.query || msg.body.query;
  const now = new Date().toISOString();
  const lastPoll = snapshot.lastPoll || new Date(0).toISOString();
  this.logger.info('Last polling timestamp=%s', lastPoll);
  const sql = originalSql.split(LAST_POLL_PLACEHOLDER).join(lastPoll);
  this.logger.debug('Transformed query is ready');
  const that = this;
  return co(function* gen() {
    const request = new cosql.Request(connection);
    request.stream = true;
    request.on('recordset', () => {
      that.logger.trace('Have got recordset metadata');
    });
    request.on('row', (row) => {
      this.emit('data', eioUtils.newMessageWithBody(row));
    });
    request.on('error', (err) => {
      this.emit('error', err);
    });
    request.on('done', (affected) => {
      that.logger.info('Query execution completed, affected=%s', affected);
      this.emit('snapshot', {
        lastPoll: now,
      });
      this.emit('end');
    });
    yield request.query(sql);
  }.bind(this));
}
function processAction ( msg , cfg , snapshot = { } ) { const originalSql = cfg . query || msg . body . query ; const now = new Date ( ) . toISOString ( ) ; const lastPoll = snapshot . lastPoll || new Date ( 0 ) . toISOString ( ) ; this . logger . info ( 'Last polling timestamp=%s' , lastPoll ) ; const sql = originalSql . split ( LAST_POLL_PLACEHOLDER ) . join ( lastPoll ) ; this . logger . debug ( 'Transformed query is ready' ) ; const that = this ; return co ( function * gen ( ) { const request = new cosql . Request ( connection ) ; request . stream = true ; request . on ( 'recordset' , ( ) => { that . logger . trace ( 'Have got recordset metadata' ) ; } ) ; request . on ( 'row' , ( row ) => { this . emit ( 'data' , eioUtils . newMessageWithBody ( row ) ) ; } ) ; request . on ( 'error' , ( err ) => { this . emit ( 'error' , err ) ; } ) ; request . on ( 'done' , ( affected ) => { that . logger . info ( 'Query execution completed, affected=%s' , affected ) ; this . emit ( 'snapshot' , { lastPoll : now , } ) ; this . emit ( 'end' ) ; } ) ; yield request . query ( sql ) ; } . bind ( this ) ) ; }
This method will be called from elastic.io platform providing following data
 @param msg incoming message object that contains ``body`` with payload
 @param cfg configuration that is account information and configuration field values
This method will be called from elastic . io platform providing following data @param msg incoming message object that contains ` ` body ` ` with payload @param cfg configuration that is account information and configuration field values
// Last poll should come from Snapshot, if not it's beginning of time // Run it