Added polling

Author: zubairov

README.md

@@ -29,6 +29,14 @@ With this action you may fetch data out of the database, e.g. using ``SELECT`` s
 This trigger & action has no limitations on the number of rows so you may expect to get all of these
 via sequential fetching that is implemented within the node.js ``mssql`` driver.
 
+#### Polling
+
+Comopnent will remember last execution timestamp and let you build queries on it:
+
+```sql
+select * from Leads where Created >= '%%EIO_LAST_POLL%%'
+```
+
 ### INSERT/DELETE/UPDATE Action
 
 ![image](https://cloud.githubusercontent.com/assets/56208/22904204/cef8cb06-f23b-11e6-998f-3fe65ab81540.png)

component.json

@@ -14,6 +14,7 @@
   "actions": {
     "insert": {
       "title": "INSERT/UPDATE/DELETE",
+      "description": "Executes a single SQL statement that causes database data update and returns a number of affected records, like INSERT, UPDATE or DELETE",
       "main": "./lib/actions/insert.js",
       "fields": {
         "query": {
@@ -28,6 +29,7 @@
     },
     "selectAction": {
       "title": "SELECT",
+      "description": "Executes a SELECT statement that fetches potentially multiple database rows from the database",
       "main": "./lib/actions/select.js",
       "fields": {
         "query": {
@@ -43,6 +45,7 @@
   "triggers": {
     "selectTrigger": {
       "title": "SELECT",
+      "description": "Executes a SELECT statement that fetches potentially multiple database rows from the database",
       "main": "./lib/actions/select.js",
       "type": "polling",
       "fields": {

lib/actions/select.js

@@ -4,6 +4,8 @@ const eioUtils = require('elasticio-node').messages;
 const co = require('co');
 const cosql = require('co-mssql');
 
+const LAST_POLL_PLACEHOLDER = "%%EIO_LAST_POLL%%";
+
 let connection;
 
 /**
@@ -28,9 +30,15 @@ function init(cfg) {
  * @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) {
-    const sql = cfg.query;
-    console.log('Executing query=%s', sql);
+function processAction(msg, cfg, snapshot) {
+    const originalSql = cfg.query;
+    const now = new Date().toISOString();
+    // Last poll should come from Snapshot, if not it's beginning of time
+    const lastPoll = snapshot?snapshot.lastPoll:null || new Date(0).toISOString();
+    console.log('Last polling timestamp=%s', lastPoll);
+    const sql = originalSql.split(LAST_POLL_PLACEHOLDER).join(lastPoll);
+    console.log('Original query=%s', originalSql);
+    console.log('Transformed query=%s', sql);
     return co(function* gen() {
         const request = new cosql.Request(connection);
         request.stream = true;
@@ -49,6 +57,9 @@ function processAction(msg, cfg) {
 
         request.on('done', (affected) => {
             console.log('Query execution completed, affected=%s', affected);
+            this.emit('snapshot', {
+                lastPoll: now
+            });
             this.emit('end');
         });
 

spec-integration/integration.spec.js

@@ -13,7 +13,10 @@ class TestEmitter extends EventEmitter {
     this.error = [];
 
     this.on('data', (value) => this.data.push(value));
-    this.on('error', (value) => this.error.push(value));
+    this.on('error', (value) => {
+        this.error.push(value)
+        console.error(value.stack || value);
+    });
     this.on('end', () => {
       this.end++;
       done();
@@ -77,7 +80,7 @@ describe('Integration test', () => {
       return select.init(cfg);
     });
 
-    it('should insert data', (done) => {
+    it('should select data', (done) => {
       const emitter = new TestEmitter(() => {
         expect(emitter.error.length).to.equal(0);
         expect(emitter.data.length).to.equal(10);
@@ -91,4 +94,28 @@ describe('Integration test', () => {
     });
   });
 
+  describe('for polling SELECT', () => {
+
+        const cfg = {
+            uri : process.env.MSSQL_URL,
+            query: "select * from Leads where Created >= '%%EIO_LAST_POLL%%'"
+        };
+
+        before(() => {
+            return select.init(cfg);
+        });
+
+        it('should insert data', (done) => {
+            const emitter = new TestEmitter(() => {
+                expect(emitter.error.length).to.equal(0);
+                expect(emitter.end).to.equal(1);
+                done();
+            });
+            const msg = {
+                body: {}
+            };
+            select.process.call(emitter, msg, cfg).catch(err => done(err));
+        });
+    });
+
 });