title: how to use nodejs to monitor oracle table data
body: <p>I am trying to find an article that shows this or something similar but i cannot. everything is based on front end. What i want to achieve is ability to check for real-time changes to a database table in oracle and for every new record that gets inserted into this table i would like to run a process for that. I have come across <a href="https://dzone.com/articles/real-time-data-with-nodejs-socketio-and-oracle-dat" rel="nofollow noreferrer">https://dzone.com/articles/real-time-data-with-nodejs-socketio-and-oracle-dat</a> but it deals with front end not exactly sure if this is the way to go.</p>
  
  <p>currently i am using cronjob using php to run my jobs but it affects the application is the <code>TNS listener</code> is getting spammed with connection requests. i need to minimize my overall application connections.</p>
  
  <p>below is what i have so far. the query only runs once. what i want it to do is run continuously until it finds a record. once it finds a record for it to run another script</p>
  
  <pre><code>var http = require('http');
  var express = require('express');
  var bodyParser = require('body-parser');
  var logger = require('morgan');
  var serveStatic = require('serve-static');
  var socketio = require('socket.io');
  var oracledb = require('oracledb');
  var dbConfig = require('./dbconfig.js');
  var app;
  var httpServer;
  var io;
  
  initWebServer();
  
  function initWebServer() {
      app = express();
      httpServer = http.Server(app);
      io = socketio(httpServer);
  
      app.use(logger('combined'));
      app.use(bodyParser.urlencoded({ extended: false }));
  
      //Create 2 static file end points, 1 for public and 1 for bower_components
      //app.use('/', serveStatic(__dirname + '/public'));
      //app.use('/vendor', serveStatic(__dirname + '/bower_components'));
  
      //Create end points for the application
      //app.get('/api/super-cities', function(req, res, next) {
          //getFavoriteCities(req, res, next, 'API');
      //});
  
      //app.post('/api/vote', postVote);
  
      //Create an end point for the database
      //app.get('/db', function(req, res, next) {
          //getFavoriteCities(req, res, next, 'SOCKET_IO');
      //});
  
      httpServer.listen(3000, function() {
          console.log('Webserver listening on localhost:3000');
          getNewCommsMatrix();
          io.on('change', function() {
              getNewCommsMatrix();
          });
      });
  }
  
  function getNewCommsMatrix(){
      oracledb.getConnection(
          dbConfig,
          function(err, connection) {
              if (err) return next(err);
  
              connection.execute(
                  'select * ' +
                    'from ( ' +
                    'SELECT wj.*, min(ts_created) over () as min_created ' +
                          'FROM CCP.worker_jobs wj WHERE flag IN (\'Q\',\'F\') AND (job = \'processCommsMatrixUpload\') ' +
                          'ORDER BY ts_created ' +
                          ') ' +
                          'where ts_created = min_created',
                  {},//no binds
                  {
                      maxRows: 1,
                      outFormat: oracledb.OBJECT
                  },
                  function(err, results) {
                      if (err) {
                          return connection.release(function() {
                              next(err);
                          });
                      }
  
                      console.log(results.metaData); 
                      console.log(results.rows);   
  
  
                      connection.release(function(err) {
                          if (err) return next(err);
  
                          /*
                          if (context === 'API') {
                              res.send(results.rows);
                          } else if (context === 'SOCKET_IO') {
                              io.emit('change', results.rows);
                              res.send();
                          }
                          */
                      });
                  }
              );
          }
      );
  }
  </code></pre>
  
  <p>my cron runs a bash script every minute which runs a process with randomn inter of 2-5s and spawns a maximum of 10.</p>
  
  <p><strong>cron</strong></p>
  
  <pre><code>CMD=$(ps aux | grep "cli-worker-processQueueCommsMatrixUpload.sh -e $ENV_NAME &gt;&gt;" | grep -v grep | wc -l)
  
  if [[ "$CMD" -lt 10 ]]; then
      COUNTER=1
  
      if [[ $ENV_NAME == test ]]; then
          while true ; do
              if [[ "$COUNTER" -eq 180 ]]; then
                  exit 0
              fi 
              /opt/SP/php7/bin/php /var/httpd/ccp-test/ccp-test.domain.com/scripts/cli-worker-manager.php processQueueCommsMatrixUpload &gt;&gt; /var/httpd/ccp-test/ccp-test.domain.com/scripts/logs/cli-worker-processQueueCommsMatrixUpload.log 2&gt;&amp;1 &amp; 
              sleep 3; 
              COUNTER=$[$COUNTER +1]
          done
      fi
  
  fi
  </code></pre>
  
  <p><strong>cli-worker-manager.php</strong></p>
  
  <pre><code>for($i=0;$i&lt;100;$i++){
          $arg = $argv[1];
          $workerManager = new WorkerManager($arg);
          $workerManager-&gt;{$arg}();
      }
  </code></pre>
  
  <p><strong>runs below function</strong></p>
  
  <pre><code>public function processQueueCommsMatrixUpload() {
          $workerJobs = new WorkerJobs;
          $dbh_ro = new DatabaseConnection;
  
          $queue = $workerJobs-&gt;getOldestJob('processCommsMatrixUpload');
          if(count($queue) &gt; 0) {
              $job = $queue[0];
              $filecache = CCPSHARED . "/locks/".$job['id'];
  
  
              $condition = [
                  'ID' =&gt; $job['id'],
                  'FLAG' =&gt; ['IN', [ 'Q', 'F' ]]
              ];
  
              if(($job['flag'] == 'F') &amp;&amp; ($job['attempts'] = 3)) {
                  $this-&gt;log-&gt;info('job #' . $job['id'] . ' Failed attempts 3 reached ');
  
                  $update = [
                      'SERVER' =&gt; gethostname(),
                      'FLAG' =&gt; 'T', // T = terminated
                      'ATTEMPTS' =&gt; 3,
                      'TS_UPDATED' =&gt; 'systimestamp',
                  ];
  
                  $result = $workerJobs-&gt;updateTable($update, $condition);
                  \unlink($filecache);
                  return true;
              }
  
              //recheck to see other worker has not processed this file
              $nextjob = $workerJobs-&gt;recheckNextJob((int) $job['id'],'processCommsMatrixUpload');
              if(count($nextjob) &gt; 0){
                  //check filecache or create file cache with
  
                  if(\file_exists($filecache)){
                      $this-&gt;log-&gt;warn('file cache exists, processCommsMatrixUpload has been picked up by other server #' . $job['id'] );
                      return true;
                  }
                  \touch($filecache);
  
                  $attempts = (empty($nextjob[0]['attempts'])) ? 1 : $nextjob[0]['attempts'];
                  $update = [
                      'SERVER' =&gt; gethostname(),
                      'FLAG' =&gt; 'P',
                      'ATTEMPTS' =&gt; $attempts,
                      'TS_UPDATED' =&gt; 'systimestamp',
                  ];
  
                  $result = $workerJobs-&gt;updateTable($update, $condition);
  
                  if(is_string($result)) {
                      $this-&gt;log-&gt;error('worker job #' . $job['id'] . ' update failure.');
  
                      $notification = new Notification();
                      $body = 'worker job #' . $job['id'] . ' update failure.';
                      $notification-&gt;cliJobs($notification-&gt;getSystemEmail(), ['user@domain.com',], $body);
                  } else {
                      sleep(rand(2 , 10)); //add random delay between each job from 1-3 secs
  
                      $ts_start = strtoupper(date('d-M-y h.i.s') ) . substr((string)microtime(), 1, 8);
                      $this-&gt;log-&gt;info('starting job #' . $job['id'] . ' at ' . $ts_start);
  
                      $this-&gt;{$job['job']}($job);
                      if(file_exists($filecache)){
                          \unlink($filecache);
                      }
                      $ts_end = strtoupper(date('d-M-y h.i.s')) . substr((string)microtime(), 1, 8);
                      $this-&gt;log-&gt;info('finishing job #' . $job['id'] . ' at ' . $ts_end);
                      $this-&gt;log-&gt;info('job #' . $job['id'] . ' processing time : '. (strtotime($ts_end) - strtotime($ts_start)) .  ' seconds');
  
                  }
              }else{
                  $this-&gt;log-&gt;warn('Job (processCommsMatrixUpload) has been picked up by other server #' . $job['id'] );
                  //$notification = new Notification();
                  //$body = 'Job (processCommsMatrixUpload) has been picked up by other server #' . $job['id'];
                  //$notification-&gt;cliJobs($notification-&gt;getSystemEmail(), ['user@domain.com'], $body);
              }
  
          }
          $workerJobs-&gt;ccpRO-&gt;disconnect();
          return true;
      }
  </code></pre>
  
  <p>i cant run as db trigger because i need it to process the query resultset in php unless there's a way to call php script from db?</p>
  
  <p>the problem is not with db connections, its with <code>TNS listener</code> which throws the error as it treats it like a <code>DDOS</code> attack.</p>
  
  <p>I thought i can have a single connection open instead of opening new connection every second.</p>