From 5b2142e409c0667fcf642d0645ce87cc32e7d8d4 Mon Sep 17 00:00:00 2001 From: Luigi Dell'Aquila Date: Thu, 26 Mar 2015 16:26:44 +0100 Subject: [PATCH 1/4] supporting live query --- lib/db/index.js | 61 +++++++++++++++++++- lib/transport/binary/connection.js | 24 +++++++- lib/transport/binary/operation-status.js | 3 +- lib/transport/binary/protocol28/operation.js | 47 +++++++++++++-- 4 files changed, 127 insertions(+), 8 deletions(-) diff --git a/lib/db/index.js b/lib/db/index.js index e7482a4..ab68fca 100644 --- a/lib/db/index.js +++ b/lib/db/index.js @@ -302,6 +302,66 @@ 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){ + console.log(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 +623,6 @@ Db.prototype.createFn = function (name, fn, options) { return this.query('CREATE FUNCTION '+name+' "'+body+'" '+params+' LANGUAGE Javascript'); }; - /** * Flatten an array of arrays */ diff --git a/lib/transport/binary/connection.js b/lib/transport/binary/connection.js index ce69149..86ea949 100644 --- a/lib/transport/binary/connection.js +++ b/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]; diff --git a/lib/transport/binary/operation-status.js b/lib/transport/binary/operation-status.js index ec971bf..c701e67 100644 --- a/lib/transport/binary/operation-status.js +++ b/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; \ No newline at end of file +exports.PUSH_DATA = 5; +exports.LIVE_RESULT = 6; \ No newline at end of file diff --git a/lib/transport/binary/protocol28/operation.js b/lib/transport/binary/protocol28/operation.js index ee959c1..1d88952 100644 --- a/lib/transport/binary/protocol28/operation.js +++ b/lib/transport/binary/protocol28/operation.js @@ -394,10 +394,49 @@ 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); + + var obj = {}; + obj.content = content; + obj.type = 'd'; + obj.cluster = clusterId; + obj.position = clusterPosition; + obj.version = version; + + return [Operation.LIVE_RESULT, token, operation, obj, 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; From 28059fa978107247e23bc3ef422c099967d94126 Mon Sep 17 00:00:00 2001 From: Luigi Dell'Aquila Date: Thu, 26 Mar 2015 16:36:14 +0100 Subject: [PATCH 2/4] fixed live query --- lib/transport/binary/protocol28/operation.js | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/transport/binary/protocol28/operation.js b/lib/transport/binary/protocol28/operation.js index 1d88952..d339ce7 100644 --- a/lib/transport/binary/protocol28/operation.js +++ b/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; From 57f9f32acd2151cc4c3113db47a0bcb1ff595ccd Mon Sep 17 00:00:00 2001 From: Luigi Dell'Aquila Date: Thu, 26 Mar 2015 16:53:08 +0100 Subject: [PATCH 3/4] removed useless console log --- lib/db/index.js | 1 - 1 file changed, 1 deletion(-) diff --git a/lib/db/index.js b/lib/db/index.js index ab68fca..1230a4c 100644 --- a/lib/db/index.js +++ b/lib/db/index.js @@ -338,7 +338,6 @@ Db.prototype.liveQuery = function (command, options) { return results; }) .then(function (response){ - console.log(response); if(response.length > 0){ var iToken = response[0].token; var parentDb = this; From 1ff1ded9c7172a65d099904c97125e0e220b9396 Mon Sep 17 00:00:00 2001 From: Luigi Dell'Aquila Date: Tue, 19 May 2015 08:26:30 +0200 Subject: [PATCH 4/4] live query - code cleanup (duplicate variable and some semicolins) --- lib/db/index.js | 4 ++-- lib/transport/binary/protocol28/operation.js | 22 +++++++++++++------- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/lib/db/index.js b/lib/db/index.js index 1230a4c..e9203ac 100644 --- a/lib/db/index.js +++ b/lib/db/index.js @@ -352,11 +352,11 @@ Db.prototype.liveQuery = function (command, options) { parentDb.emit("live-create", result); } } - } + }; this.server.transport.connection.on("live-query-result", wrapperCallback); } - }) + }); return this; }; diff --git a/lib/transport/binary/protocol28/operation.js b/lib/transport/binary/protocol28/operation.js index d339ce7..da9f66a 100644 --- a/lib/transport/binary/protocol28/operation.js +++ b/lib/transport/binary/protocol28/operation.js @@ -425,14 +425,20 @@ Operation.prototype.consume = function (buffer, offset) { offset += contentLenght; var content = deserializer.deserialize(asString, this.data.transformerFunctions); - var obj = {}; - obj.content = content; - obj.type = 'd'; - obj.cluster = clusterId; - obj.position = clusterPosition; - obj.version = version; - - return [Operation.LIVE_RESULT, token, operation, obj, offset]; + 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 = {};