Livequery

lib/db/index.js

@@ -302,6 +302,65 @@ Db.prototype.query = function (command, options) {
   });
 };
 
+/**
+ * Execute a live query against the database
+ *
+ * @param   {String} query   The query or command to execute.
+ * @param   {Object} options The options for the query / command.
+ * @promise {Mixed}          The token of the live query.
+ */
+Db.prototype.liveQuery = function (command, options) {
+  options = options || {};
+  options.mode = 'l';
+  options.class='q';
+  this.exec(command, options)
+      .bind(this)
+      .then(function (response) {
+        if (!response.results || response.results.length === 0) {
+          return [[], []];
+        }
+        return response.results
+            .map(this.normalizeResult, this)
+            .reduce(flatten, [])
+            .reduce(function (list, item) {
+              if (item && item['@preloaded']) {
+                delete item['@preloaded'];
+                list[1].push(item);
+              }
+              else {
+                list[0].push(item);
+              }
+              return list;
+            }, [[], []]);
+      })
+      .spread(function (results, preloaded) {
+        this.record.resolveReferences(results.concat(preloaded));
+        return results;
+      })
+      .then(function (response){
+        if(response.length > 0){
+          var iToken = response[0].token;
+          var parentDb = this;
+          var wrapperCallback = function(currentToken, operation, result){
+            if(currentToken == iToken) {
+              if (operation === 1) {
+                parentDb.emit("live-update", result);
+              } else if (operation === 2) {
+                parentDb.emit("live-delete", result);
+              } else if (operation === 3) {
+                parentDb.emit("live-insert", result);
+                parentDb.emit("live-create", result);
+              }
+            }
+          };
+          this.server.transport.connection.on("live-query-result", wrapperCallback);
+
+        }
+      });
+  return this;
+};
+
+
 /**
  * Normalize a result, where possible.
  * @param  {Object} result The result to normalize.
@@ -563,7 +622,6 @@ Db.prototype.createFn = function (name, fn, options) {
   return this.query('CREATE FUNCTION '+name+' "'+body+'" '+params+' LANGUAGE Javascript');
 };
 
-
 /**
  * Flatten an array of arrays
  */

lib/transport/binary/connection.js

@@ -6,7 +6,9 @@ var net = require('net'),
     errors = require('../../errors'),
     OperationStatus = require('./operation-status'),
     EventEmitter = require('events').EventEmitter,
-    Promise = require('bluebird');
+    Promise = require('bluebird'),
+    Operation = require('./protocol28/operation'); //TODO refactor this!!!
+
 
 function Connection (config) {
   EventEmitter.call(this);
@@ -352,8 +354,26 @@ Connection.prototype.destroySocket = function () {
  * @return {Integer}        The offset that was successfully read up to.
  */
 Connection.prototype.process = function (buffer, offset) {
-  var code, parsed, result, status, item, op, deferred, err;
+  var code, parsed, result, status, item, op, deferred, err, token, operation;
   offset = offset || 0;
+  if(this.queue.length === 0){
+    op = new Operation();//TODO refactor this!
+    parsed = op.consume(buffer, offset);
+    status = parsed[0];
+    if (status === OperationStatus.PUSH_DATA) {
+      offset = parsed[1];
+      result = parsed[2];
+      this.emit('update-config', result);
+      return offset;
+    }else if(status === OperationStatus.LIVE_RESULT){
+      token = parsed[1];
+      operation = parsed[2];
+      result = parsed[3];
+      offset = parsed[4];
+      this.emit('live-query-result', token, operation, result);
+      return offset;
+    }
+  }
   while ((item = this.queue.shift())) {
     op = item[0];
     deferred = item[1];

lib/transport/binary/operation-status.js

@@ -5,4 +5,5 @@ exports.WRITTEN = 1;
 exports.READING = 2;
 exports.COMPLETE = 3;
 exports.ERROR = 4;
-exports.PUSH_DATA = 5;
+exports.PUSH_DATA = 5;
+exports.LIVE_RESULT = 6;

lib/transport/binary/protocol28/operation.js

@@ -38,6 +38,8 @@ Operation.READING = statuses.READING;
 Operation.COMPLETE = statuses.COMPLETE;
 Operation.ERROR = statuses.ERROR;
 Operation.PUSH_DATA = statuses.PUSH_DATA;
+Operation.LIVE_RESULT = statuses.LIVE_RESULT;
+
 
 // make it easy to inherit from the base class
 Operation.extend = utils.extend;
@@ -394,10 +396,55 @@ Operation.prototype.consume = function (buffer, offset) {
 
     code = buffer.readUInt8(offset);
     if (code === 3) {
-      offset += 5; // ignore the next integer
-      obj = {};
-      offset += this.parsePushedData(buffer, offset, obj, 'data');
-      return [Operation.PUSH_DATA, offset, obj.data];
+      var sessionId = buffer.readInt32BE(offset+1);
+      var pushType = buffer.readUInt8(offset + 5);
+
+      if (pushType === 81) {
+        offset += 6; // ignore the next integer
+        var length = buffer.readInt32BE(offset);
+        offset+=4;
+        var operation = buffer.readUInt8(offset);
+        offset+=1;
+        var token = buffer.readInt32BE(offset);
+        offset+=4;
+        var recordType = buffer.readUInt8(offset);
+        offset+=1;
+        var version = buffer.readInt32BE(offset);
+        offset+=4;
+        var clusterId = buffer.readInt16BE(offset);
+        offset+=2;
+        var clusterPosition = Long.fromBits(
+            buffer.readUInt32BE(offset + 4),
+            buffer.readInt32BE(offset)
+        ).toNumber();
+        offset+=8;
+        var contentLenght = buffer.readInt32BE(offset);
+        offset+=4;
+
+        var asString = buffer.toString('utf8', offset, offset + contentLenght);
+        offset += contentLenght;
+        var content = deserializer.deserialize(asString, this.data.transformerFunctions);
+
+        return [
+          Operation.LIVE_RESULT,
+          token,
+          operation,
+          {
+            content: content,
+            type: 'd',
+            cluster: clusterId,
+            position: clusterPosition,
+            version: version
+          },
+          offset
+        ];
+
+      } else {
+        offset += 5; // ignore the next integer
+        obj = {};
+        offset += this.parsePushedData(buffer, offset, obj, 'data');
+        return [Operation.PUSH_DATA, offset, obj.data];
+      }
     }
 
     this.status = Operation.READING;