lib/triggers/getDataPolling.js
/**
 * This method will be called from OIH 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
 * @param snapshot - saves the current state of integration step for the future reference
 */
This method will be called from OIH 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
@param snapshot - saves the current state of integration step for the future reference
This method will be called from OIH 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 @param snapshot - saves the current state of integration step for the future reference
async function processTrigger(msg, cfg, snapshot = {}) {
  try {
    const {
      databaseType
    } = cfg;
    console.dir(msg);
    console.dir( cfg);
    try{
    if ('query' in msg.data){
      query=msg.data.query;}
    else {
      query=cfg.query;
    }}catch (e) {
    console.log(`ERROR: ${e}`);
    }
    snapshot.lastUpdated = snapshot.lastUpdated || 1;
    async function handleResult(data, last) {
      console.log("INFO: entered handleResults. data=" + data + " last=" + last);
      if (data === false) {
        if (cfg.devMode) console.log('Skipping empty entry');
      } else if (data === true) {
        snapshot.lastUpdated = parseInt(last.sequenceNumber, 10) + 1;
        snapshot.date = last.date;
        if (cfg.devMode) console.log(`New snapshot: ${JSON.stringify(snapshot, undefined, 2)}`);
        this.emit('snapshot', snapshot);
      } else {
        const newElement = { meta: data.recordUid };
        const transformedData = transform(data, cfg);
        newElement.data = transformedData;
        console.log("INFO: handleResults.newElement.data=" + newElement);
        if (cfg.devMode) console.log('newElement', newElement);
        this.emit('data', newElement);
      }
    }
    const matches = query.match(/(^|[\s)(]+)(delete|insert|update|set)(?=\s|\(|\)|$)/iu);
    if (matches !== null) {
      console.error('SQL edit operation found --> Aborting fetch');
      console.error('Query was:', query);
    }
    if (databaseType) {
      const dbType = databaseType.toLowerCase();
      if (dbType === 'mysql') {
        mysqlFetch(cfg, query, handleResult.bind(this));
      } else if (dbType === 'postgresql') {
        postgreFetch(cfg, query, handleResult.bind(this));
      } else if (dbType === 'mssql' || dbType === 'sql') {
        mssqlFetch(cfg, query, handleResult.bind(this));
      } else {
        console.error(`Unknown database type [${databaseType}] falling back to MySQL`);
        mysqlFetch(cfg, query, handleResult.bind(this));
      }
    } else {
      console.error('No database type set asuming MySQL');
      mysqlFetch(cfg, query, handleResult.bind(this));
    }
    console.log('Finished execution');
    this.emit('end');
  } catch (e) {
    console.log(`ERROR: ${e}`);
    this.emit('error', e);
  }
}
async function processTrigger ( msg , cfg , snapshot = { } ) { try { const { databaseType } = cfg ; console . dir ( msg ) ; console . dir ( cfg ) ; try { if ( 'query' in msg . data ) { query = msg . data . query ; } else { query = cfg . query ; } } catch ( e ) { console . log ( ` ${ e } ` ) ; } snapshot . lastUpdated = snapshot . lastUpdated || 1 ; async function handleResult ( data , last ) { console . log ( "INFO: entered handleResults. data=" + data + " last=" + last ) ; if ( data === false ) { if ( cfg . devMode ) console . log ( 'Skipping empty entry' ) ; } else if ( data === true ) { snapshot . lastUpdated = parseInt ( last . sequenceNumber , 10 ) + 1 ; snapshot . date = last . date ; if ( cfg . devMode ) console . log ( ` ${ JSON . stringify ( snapshot , undefined , 2 ) } ` ) ; this . emit ( 'snapshot' , snapshot ) ; } else { const newElement = { meta : data . recordUid } ; const transformedData = transform ( data , cfg ) ; newElement . data = transformedData ; console . log ( "INFO: handleResults.newElement.data=" + newElement ) ; if ( cfg . devMode ) console . log ( 'newElement' , newElement ) ; this . emit ( 'data' , newElement ) ; } } const matches = query . match ( / (^|[\s)(]+)(delete|insert|update|set)(?=\s|\(|\)|$) / iu ) ; if ( matches !== null ) { console . error ( 'SQL edit operation found --> Aborting fetch' ) ; console . error ( 'Query was:' , query ) ; } if ( databaseType ) { const dbType = databaseType . toLowerCase ( ) ; if ( dbType === 'mysql' ) { mysqlFetch ( cfg , query , handleResult . bind ( this ) ) ; } else if ( dbType === 'postgresql' ) { postgreFetch ( cfg , query , handleResult . bind ( this ) ) ; } else if ( dbType === 'mssql' || dbType === 'sql' ) { mssqlFetch ( cfg , query , handleResult . bind ( this ) ) ; } else { console . error ( ` ${ databaseType } ` ) ; mysqlFetch ( cfg , query , handleResult . bind ( this ) ) ; } } else { console . error ( 'No database type set asuming MySQL' ) ; mysqlFetch ( cfg , query , handleResult . bind ( this ) ) ; } console . log ( 'Finished execution' ) ; this . emit ( 'end' ) ; } catch ( e ) { console . log ( ` ${ e } ` ) ; this . emit ( 'error' , e ) ; } }
This method will be called from OIH 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
 @param snapshot - saves the current state of integration step for the future reference
This method will be called from OIH 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 @param snapshot - saves the current state of integration step for the future reference
// Set the snapshot if it is not provided // @todo: ! // All done let's make a snapshot // Prepare and emit entry // Emit the object with meta and data properties