Permalink
Browse files

Finish v0.8.11

  • Loading branch information...
itayw committed Jun 22, 2015
2 parents cbe8e02 + f9b1558 commit 6cc9317114189094fecbb7692d957428a780f407
View

This file was deleted.

Oops, something went wrong.
View
@@ -11,10 +11,14 @@
'use strict';
var joola = require('../joola');
try {
module.exports = require('geoip-lite');
joola.logger.info('Found geoip-lite module.');
}
catch (ex) {
joola.logger.warn('Could not find geoip-lite module.');
exports.lookup = function (ip) {
return ip;
};
View
@@ -26,36 +26,86 @@ var
var etl = {};
exports.etl = etl;
Array.prototype.clean = function (deleteValue) {
for (var i = 0; i < this.length; i++) {
if (!this[i]) {
this.splice(i, 1);
i--;
}
}
return this;
};
etl.verify = function (context, workspace, collection, documents, callback) {
var _document;
if (!Array.isArray(documents))
documents = [documents];
documents.clean();
var buildRowKey = function (obj) {
var key = obj.timestamp.toISOString();
var key;
try {
key = obj.timestamp.toISOString();
}
catch (ex) {
key = '';
}
traverse.map(obj, function (x) {
if (x && typeof (x) === 'string') {
key += x;
}
});
key = joola.common.hash(key);
return key;
};
async.map(documents, function (doc, callback) {
async.map(documents, function (doc, cb) {
try {
if (!doc.timestamp)
doc.timestamp = new Date();
else {
doc.timestamp = new Date(doc.timestamp.value || doc.timestamp);
}
doc._key = buildRowKey(doc);
Object.keys(doc).forEach(function (key) {
var elem = doc[key];
try {
if (elem.datatype === 'ip' || elem.key === 'ip' || key === 'ip') {
var ipaddress = elem.value || elem;
var ip = joola.common.geoip.lookup(ipaddress);
if (ip && typeof ip === 'object') {
delete ip.range;
ip.address = ipaddress;
doc[key] = ip;
doc.location = {
lat: ip.ll[0],
lon: ip.ll[1]
};
}
else {
ip = {};
ip.address = ipaddress;
doc[key] = ip;
doc.location = {
lat: 0,
lon: 0
};
}
}
}
catch (ex) {
//ignore
console.log('exception',ex);
}
});
}
catch (ex) {
return callback(ex);
doc.timestamp = new Date();
return cb(ex);
}
return callback(null);
doc._key = buildRowKey(doc);
return cb(null);
}, function (err) {
if (err)
return callback(err);
@@ -65,9 +115,7 @@ etl.verify = function (context, workspace, collection, documents, callback) {
joola.dispatch.collections.metadata(context, workspace, null, _document, function (err, meta) {
/* istanbul ignore if */
if (err)
return setImmediate(function () {
return callback(err);
});
return callback(err);
joola.dispatch.collections.get(context, workspace, collection, function (err, _collection) {
if (err) {
@@ -77,30 +125,22 @@ etl.verify = function (context, workspace, collection, documents, callback) {
joola.dispatch.collections.add(context, workspace, meta, function (err, _collection) {
/* istanbul ignore if */
if (err && err.message != 'Collection already exist') {
return setImmediate(function () {
return callback(err);
});
return callback(err);
}
joola.dispatch.collections.get(context, workspace, collection, function (err, _collection) {
if (err) {
return setImmediate(function () {
return callback(err);
});
return callback(err);
}
if (joola.datastore.providers.default.addcollection) {
joola.datastore.providers.default.addcollection(meta.key, meta, function (err) {
if (err)
return callback(err);
return setImmediate(function () {
return callback(null);
});
return callback(null);
});
}
else {
return setImmediate(function () {
return callback(null);
});
return callback(null);
}
});
});
@@ -109,9 +149,7 @@ etl.verify = function (context, workspace, collection, documents, callback) {
joola.dispatch.collections.metadata(context, workspace, collection, _document, function (err, _meta) {
/* istanbul ignore if */
if (err)
return setImmediate(function () {
return callback(err);
});
return callback(err);
var differences;
delete meta._key;
@@ -147,22 +185,16 @@ etl.verify = function (context, workspace, collection, documents, callback) {
joola.datastore.providers.default.altercollection(meta.key, meta, differences, function (err) {
if (err)
return callback(err);
return setImmediate(function () {
return callback(null);
});
return callback(null);
});
}
else {
return setImmediate(function () {
return callback(null);
});
return callback(null);
}
});
}
else {
return setImmediate(function () {
return callback(null);
});
return callback(null);
}
});
}
@@ -174,46 +206,57 @@ etl.verify = function (context, workspace, collection, documents, callback) {
etl.load = function (context, workspace, collection, documents, options, callback) {
joola.dispatch.collections.metadata(context, workspace, collection, ce.clone(documents[0]), function (err, meta, _collection) {
if (err)
return setImmediate(function () {
return callback(err);
});
return callback(err);
_collection.meta = meta;
_collection.storeKey = (workspace + '_' + collection).replace(/[^\w\s]/gi, '');
//cleanup the object if we passed adhoc values;
documents.forEach(function (document) {
traverse.forEach(document, function (x) {
if (x && typeof x === 'object' && x.hasOwnProperty('value'))
this.update(x.value);
});
});
var d = domain.create();
d.on('error', function (err) {
console.log(err);
joola.logger.error('Failed to insert document, ' + err);
return callback('Failed to insert document: ' + err);
});
d.run(function () {
var filtered = _.filter(Object.keys(joola.datastore.providers), function (key) {
var p = joola.datastore.providers[key];
return key !== 'default' && (p.enabled || !p.hasOwnProperty('enabled'));
var docCount = 0;
var traverseFound = false;
var geoFound = false;
async.map(documents, function (document, cb) {
if (docCount === 0 || (docCount > 0 && (traverseFound || geoFound))) {
traverse.forEach(document, function (x) {
if (x && typeof x === 'object' && x.hasOwnProperty('value')) {
traverseFound = true;
this.update(x.value);
}
});
docCount++;
}
return cb(null);
}, function () {
var d = domain.create();
d.on('error', function (err) {
console.log(err);
joola.logger.error('Failed to insert document, ' + err);
return callback('Failed to insert document: ' + err);
});
async.map(filtered, function (key, cb) {
var provider = joola.datastore.providers[key];
provider.insert(_collection, ce.clone(documents), options, function (err, results) {
if (err)
return cb(err);
return cb(null, results);
d.run(function () {
var filtered = _.filter(Object.keys(joola.datastore.providers), function (key) {
var p = joola.datastore.providers[key];
return key !== 'default' && (p.enabled || !p.hasOwnProperty('enabled'));
});
}, function (err, results) {
if (err)
return callback(err);
documents.forEach(function (d) {
d.saved = true;
async.map(filtered, function (key, cb) {
var provider = joola.datastore.providers[key];
provider.insert(_collection, ce.clone(documents), options, function (err, results) {
if (err)
return cb(err);
return cb(null, results);
});
}, function (err, results) {
if (err)
return callback(err);
async.map(documents, function (d, cb) {
d.saved = true;
return cb(null, d);
}, function (err, arrdocuments) {
if (err)
return callback(err);
return callback(null, arrdocuments, results);
});
});
return callback(null, documents, results);
});
});
});
@@ -305,7 +348,7 @@ exports.insert = {
}
callback = callback || function () {
};
};
if (context.user.workspace !== workspace && context.user.permissions.indexOf('superuser') === -1) {
var err = new Error('Forbidden');
err.code = 403;
@@ -327,24 +370,22 @@ exports.insert = {
if (err)
return callback(err);
return setImmediate(function () {
var end_ts = new Date().getTime();
if (count > 1)
joola.logger.trace('Beacon insert, count: ' + count + ', total: ' + (end_ts - start_ts) + 'ms, rate: ' + (count / (end_ts - start_ts)) + 'doc/ms');
var end_ts = new Date().getTime();
if (count > 1)
joola.logger.trace('Beacon insert, count: ' + count + ', total: ' + (end_ts - start_ts) + 'ms, rate: ' + (count / (end_ts - start_ts)) + 'doc/ms');
//if (workspace != '_stats')
//if (workspace != '_stats')
// joola.stats.emit({event: 'writes', workspace: workspace, username: context.user.username, collection: collection, writeCount: count, duration_per_doc: (count / (end_ts - start_ts))});
subscribers.forEach(function (subscriber, i) {
if (subscriber.disconnected)
subscribers.splice(i, 1);
else {
subscriber.emit('event', collection, documents);
}
});
return callback(null, documents);
subscribers.forEach(function (subscriber, i) {
if (subscriber.disconnected)
subscribers.splice(i, 1);
else {
subscriber.emit('event', collection, documents);
}
});
return callback(null, documents);
});
});
}
@@ -376,12 +417,13 @@ exports.subscribe = {
},
_route: function (req, res, next) {
subscribers.push(req.socket);
return router.responseSuccess({}, {}, req, res);
},
run: function (context, callback) {
callback = callback || function () {
};
};
//subscribers.push(req.socket);
return callback();
return callback(null, {});
}
};
@@ -198,7 +198,7 @@ exports.get = {
*
* ```js
* var newCollection = {
* id: 'dummyCollection',
* id: 'dummyCollection',
* name: 'Dummy Collection',
* description: 'Dummy Collection for docs',
* type: 'data',
@@ -299,7 +299,7 @@ exports.add = {
*
* ```js
* var collectionToUpdate = {
* id: 'dummyCollection',
* id: 'dummyCollection',
* name: 'Dummy Collection - updated',
* description: 'Dummy Collection for docs - updated',
* };
@@ -743,4 +743,4 @@ exports.metadata = {
return callback(null, ce.clone(_shadow));
}
}
};
};
@@ -78,6 +78,11 @@ var proto = {
"name": "onDraw",
"description": "The onDraw of the canvas",
"type": "object"
},
"overlay": {
"name": "overlay",
"description": "The overlay of the canvas",
"type": "object"
}
};
Oops, something went wrong.

0 comments on commit 6cc9317

Please sign in to comment.