From 624cf01270efe06e1fa451189ae84c459332e170 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Mon, 8 May 2017 00:44:08 -0700 Subject: [PATCH] Support caching in MongoDB instead of Redis --- .gitignore | 1 - lib/api-functions.coffee | 8 +- lib/cache.coffee | 310 ----------------------------------- lib/cache.js | 257 +++++++++++++++++++++++++++++ lib/cwmp.coffee | 190 +++++++++++++++++----- lib/db.coffee | 147 +++++------------ lib/local-cache.js | 339 +++++++++++++++++++++++++++++++++++++++ lib/nbi.coffee | 41 +++-- lib/server.coffee | 8 +- lib/session.coffee | 49 +----- 10 files changed, 815 insertions(+), 535 deletions(-) delete mode 100644 lib/cache.coffee create mode 100644 lib/cache.js create mode 100644 lib/local-cache.js diff --git a/.gitignore b/.gitignore index a5de673..3364eab 100644 --- a/.gitignore +++ b/.gitignore @@ -2,7 +2,6 @@ lib/*.map lib/api-functions.js lib/auth.js -lib/cache.js lib/common.js lib/config.js lib/cwmp.js diff --git a/lib/api-functions.coffee b/lib/api-functions.coffee index 41b019f..be8d26a 100644 --- a/lib/api-functions.coffee +++ b/lib/api-functions.coffee @@ -27,6 +27,7 @@ common = require './common' util = require 'util' URL = require 'url' auth = require './auth' +cache = require './cache' udpConReq = (address, un, key, callback) -> @@ -242,12 +243,7 @@ deleteDevice = (deviceId, callback) -> return callback(err) if err db.operationsCollection.remove({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}, (err) -> return callback(err) if err - db.redisClient.del("#{deviceId}_presets_hash", - "#{deviceId}_inform_hash", - "#{deviceId}_faults", - "#{deviceId}_tasks", - "#{deviceId}_operations", - callback) + cache.del("#{deviceId}_tasks_faults_operations", callback) ) ) ) diff --git a/lib/cache.coffee b/lib/cache.coffee deleted file mode 100644 index d8e9a3d..0000000 --- a/lib/cache.coffee +++ /dev/null @@ -1,310 +0,0 @@ -### -# Copyright 2013-2017 Zaid Abdulla -# -# This file is part of GenieACS. -# -# GenieACS is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# GenieACS is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with GenieACS. If not, see . -### - -vm = require 'vm' -crypto = require 'crypto' - -later = require 'later' - -db = require './db' -query = require './query' - - -REFRESH = 3000 - -nextRefresh = 0 -hash = null -presets = null -provisions = null -virtualParameters = null -files = null - - -UNLOCK_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end' -EXTEND_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("pexpire", KEYS[1], ARGV[2]) else return 0 end' - -lock = (lockName, ttl, callback) -> - token = Math.random().toString(36).slice(2) - unlockOrExtend = (ttl) -> - if not ttl - db.redisClient.eval(UNLOCK_SCRIPT, 1, lockName, token, (err, res) -> - throw err or new Error('Lock expired') if err or not res - ) - else - db.redisClient.eval(EXTEND_SCRIPT, 1, lockName, token, ttl, (err, res) -> - throw err or new Error('Lock expired') if err or not res - ) - return - - db.redisClient.set(lockName, token, 'NX', 'PX', ttl, (err, res) -> - if err or res - return callback(err, unlockOrExtend) - - setTimeout(() -> - return lock(lockName, ttl, callback) - , 200) - ) - - -computeHash = () -> - # MD5 hash for presets, provisions, virtual parameters for detecting changes - h = crypto.createHash('md5') - for p in presets - h.update(JSON.stringify(p.name)) - h.update(JSON.stringify(p.channel)) - h.update(JSON.stringify(p.schedule)) - h.update(JSON.stringify(p.events)) - h.update(JSON.stringify(p.precondition)) - h.update(JSON.stringify(p.provisions)) - - keys = Object.keys(provisions).sort() - h.update(JSON.stringify(keys)) - h.update(provisions[k].md5) for k in keys - - keys = Object.keys(virtualParameters).sort() - h.update(JSON.stringify(keys)) - h.update(virtualParameters[k].md5) for k in keys - - hash = h.digest('hex') - - -refresh = (callback) -> - now = Date.now() - db.redisClient.get('presets_hash', (err, res) -> - return callback(err) if err - - if hash? and res == hash - nextRefresh = now + REFRESH - return callback() - - lock('presets_hash_lock', 3000, (err, unlockOrExtend) -> - return callback(err) if err - counter = 3 - - counter += 2 - db.presetsCollection.find().toArray((err, res) -> - if err - callback(err) if counter & 1 - return counter = 0 - - db.objectsCollection.find().toArray((err, objects) -> - if err - callback(err) if (counter & 1) - return counter = 0 - - res.sort((a, b) -> - if a.weight == b.weight - return a._id > b._id - else - return a.weight - b.weight - ) - - presets = [] - - for preset in res - schedule = null - if preset.schedule - parts = preset.schedule.trim().split(/\s+/) - schedule = { - md5: crypto.createHash('md5').update(preset.schedule).digest('hex') - } - - try - schedule.duration = +(parts.shift()) * 1000 - parts.unshift('0') if parts.length == 5 - # TODO later.js doesn't throw erorr if expression is invalid! - schedule.schedule = later.schedule(later.parse.cron(parts.join(' '), true)) - catch err - # TODO show a warning - schedule.schedule = false - - events = preset.events ? {} - - precondition = query.convertMongoQueryToFilters(JSON.parse(preset.precondition)) - - _provisions = preset.provisions or [] - - # Generate provisions from the old configuration format - for c in preset.configurations - switch c.type - when 'age' - _provisions.push(['refresh', c.name, c.age]) - when 'value' - _provisions.push(['value', c.name, c.value]) - when 'add_tag' - _provisions.push(['tag', c.tag, true]) - when 'delete_tag' - _provisions.push(['tag', c.tag, false]) - when 'provision' - _provisions.push([c.name].concat(c.args or [])) - when 'add_object' - for obj in objects - if obj['_id'] == c.object - alias = ("#{k}:#{JSON.stringify(obj[k])}" for k in obj['_keys']).join(',') - p = "#{c.name}.[#{alias}]" - _provisions.push(['instances', p, 1]) - for k of obj - if k[0] != '_' and k not in obj['_keys'] - _provisions.push(['value', "#{p}.#{k}", obj[k]]) - when 'delete_object' - for obj in objects - if obj['_id'] == c.object - alias = ("#{k}:#{JSON.stringify(obj[k])}" for k in obj['_keys']).join(',') - p = "#{c.name}.[#{alias}]" - _provisions.push(['instances', p, 0]) - else - callback(new Error("Unknown configuration type #{c.type}")) if counter & 1 - return counter = 0 - - presets.push({name: preset._id, channel: preset.channel or 'default', schedule: schedule, events: events, precondition: precondition, provisions: _provisions}) - - if (counter -= 2) == 1 - computeHash() - db.redisClient.setex("presets_hash", 300, hash, (err) -> - unlockOrExtend(0) - nextRefresh = now + REFRESH - return callback() - ) - ) - ) - - counter += 2 - db.provisionsCollection.find().toArray((err, res) -> - if err - callback(err) if counter & 1 - return counter = 0 - - provisions = {} - for r in res - provisions[r._id] = {} - provisions[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex') - provisions[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: -1, timeout: 50}) - - if (counter -= 2) == 1 - computeHash() - db.redisClient.setex("presets_hash", 300, hash, (err) -> - unlockOrExtend(0) - nextRefresh = now + REFRESH - return callback() - ) - ) - - counter += 2 - db.virtualParametersCollection.find().toArray((err, res) -> - if err - callback(err) if counter & 1 - return counter = 0 - - virtualParameters = {} - for r in res - virtualParameters[r._id] = {} - virtualParameters[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex') - virtualParameters[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: -1, timeout: 50}) - - if (counter -= 2) == 1 - computeHash() - db.redisClient.setex("presets_hash", 300, hash, (err) -> - unlockOrExtend(0) - nextRefresh = now + REFRESH - return callback() - ) - ) - - counter += 2 - db.filesCollection.find().toArray((err, res) -> - if err - callback(err) if counter & 1 - return counter = 0 - - files = {} - for r in res - id = r.filename or r._id.toString() - files[id] = {} - files[id].length = r.length - files[id].md5 = r.md5 - files[id].contentType = r.contentType - - if (counter -= 2) == 1 - computeHash() - db.redisClient.setex("presets_hash", 300, hash, (err) -> - unlockOrExtend(0) - nextRefresh = now + REFRESH - return callback() - ) - ) - - if (counter -= 2) == 1 - computeHash() - db.redisClient.setex("presets_hash", 300, hash, (err) -> - unlockOrExtend(0) - nextRefresh = now + REFRESH - return callback() - ) - ) - ) - - -getPresets = (callback) -> - if Date.now() < nextRefresh - return callback(null, hash, presets) - - refresh((err) -> - return callback(err, hash, presets) - ) - - -getProvisions = (callback) -> - if Date.now() < nextRefresh - return callback(null, hash, provisions) - - refresh((err) -> - return callback(err, hash, provisions) - ) - - -getVirtualParameters = (callback) -> - if Date.now() < nextRefresh - return callback(null, hash, virtualParameters) - - refresh((err) -> - return callback(err, hash, virtualParameters) - ) - - -getFiles = (callback) -> - if Date.now() < nextRefresh - return callback(null, hash, files) - - refresh((err) -> - return callback(err, hash, files) - ) - - -getProvisionsAndVirtualParameters = (callback) -> - if Date.now() < nextRefresh - return callback(null, hash, provisions, virtualParameters) - - refresh((err) -> - return callback(err, hash, provisions, virtualParameters) - ) - -exports.getPresets = getPresets -exports.getProvisions = getProvisions -exports.getFiles = getFiles -exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters diff --git a/lib/cache.js b/lib/cache.js new file mode 100644 index 0000000..d5446cc --- /dev/null +++ b/lib/cache.js @@ -0,0 +1,257 @@ +/** + * Copyright 2013-2017 Zaid Abdulla + * + * This file is part of GenieACS. + * + * GenieACS is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * GenieACS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with GenieACS. If not, see . + */ +"use strict"; + +const config = require("./config"); + +const MAX_CACHE_TTL = config.get("MAX_CACHE_TTL"); + +let redisClient; +let mongoDb; +let mongoCollection; +let mongoTimeOffset = 0; + + +function connect(callback) { + const REDIS_HOST = config.get("REDIS_HOST"); + + if (REDIS_HOST) { + const redis = require("redis"); + const REDIS_PORT = config.get("REDIS_PORT"); + const REDIS_DB = config.get("REDIS_DB"); + redisClient = redis.createClient(REDIS_PORT, REDIS_HOST); + redisClient.select(REDIS_DB, function(err) { + exports.get = redisGet; + exports.set = redisSet; + exports.del = redisDel; + exports.pop = redisPop; + exports.lock = redisLock; + callback(err); + }); + } + else { + const mongodb = require("mongodb"); + const MONGODB_CONNECTION_URL = config.get("MONGODB_CONNECTION_URL"); + + mongodb.MongoClient.connect(MONGODB_CONNECTION_URL, function(err, db) { + if (err) + return callback(err); + + mongoDb = db; + mongoCollection = db.collection("cache"); + mongoCollection.ensureIndex({"expire": 1}, {expireAfterSeconds: 0}); + + const now = Date.now(); + mongoDb.command({hostInfo: 1}, function(err, res) { + if (err) + return callback(err); + mongoTimeOffset = res.system.currentTime.getTime() - now; + exports.get = mongoGet; + exports.set = mongoSet; + exports.del = mongoDel; + exports.pop = mongoPop; + exports.lock = mongoLock; + callback(); + }); + }); + } +} + + +function disconnect() { + if (redisClient) + redisClient.quit(); + + if (mongoDb) + mongoDb.close(); +} + + +function redisGet(key, callback) { + if (Array.isArray(key)) + redisClient.mget(key, callback); + else + redisClient.get(key, callback); +} + + +function redisDel(key, callback) { + redisClient.del(key, callback); +} + + +function redisSet(key, value, ttl, callback) { + if (!callback && typeof v === "function") { + callback = ttl; + ttl = null; + } + + if (!ttl) + ttl = MAX_CACHE_TTL; + + redisClient.setex(key, ttl, value, callback); +} + + +function redisPop(key, callback) { + const POP_SCRIPT = "local v=redis.call(\"get\",KEYS[1]);redis.call(\"del\",KEYS[1]);return v;"; + redisClient.eval(POP_SCRIPT, 1, key, callback); +} + + +function redisLock(lockName, ttl, callback) { + const UNLOCK_SCRIPT = "if redis.call(\"get\", KEYS[1]) == ARGV[1] then return redis.call(\"del\", KEYS[1]) else return 0 end"; + const EXTEND_SCRIPT = "if redis.call(\"get\", KEYS[1]) == ARGV[1] then return redis.call(\"pexpire\", KEYS[1], ARGV[2]) else return 0 end"; + const token = Math.random().toString(36).slice(2); + + function unlockOrExtend(ttl) { + if (!ttl) { + redisClient.eval(UNLOCK_SCRIPT, 1, lockName, token, function(err, res) { + if (err || !res) + throw err || new Error("Lock expired"); + }); + } + else { + redisClient.eval(EXTEND_SCRIPT, 1, lockName, token, ttl, function(err, res) { + if (err || !res) + throw err || new Error("Lock expired"); + }); + } + } + + redisClient.set(lockName, token, "NX", "PX", ttl, function(err, res) { + if (err || res) + return callback(err, unlockOrExtend); + + setTimeout(function() { + redisLock(lockName, ttl, callback); + }, 200); + }); +} + + +function mongoGet(key, callback) { + let expire = new Date(Date.now() - mongoTimeOffset); + if (Array.isArray(key)) + mongoCollection.find({"_id": {"$in": key}}).toArray(function(err, res) { + if (err) + return callback(err); + + let indices = {}; + key.forEach(function(v, i) { + indices[v] = i; + }); + + let values = []; + res.forEach(function(r) { + if (r["expire"] > expire) + values[indices[r["_id"]]] = r["value"]; + }); + callback(null, values); + }); + else + mongoCollection.findOne({"_id": {"$in": [key]}}, function(err, res) { + if (err || !res) + return callback(err); + + if (res["expire"] > expire) + return callback(null, res["value"]); + + callback(); + }); +} + + +function mongoDel(key, callback) { + if (Array.isArray(key)) + mongoCollection.remove({"_id": {"$in": key}}, callback); + else + mongoCollection.remove({"_id": key}, callback); +} + + +function mongoSet(key, value, ttl, callback) { + if (!callback && typeof v === "function") { + callback = ttl; + ttl = null; + } + + if (!ttl) + ttl = MAX_CACHE_TTL; + + let expire = new Date((Date.now() - mongoTimeOffset) + (ttl * 1000)); + mongoCollection.save({"_id": key, "value": value, "expire": expire}, callback); +} + + +function mongoPop(key, callback) { + mongoCollection.findAndModify({"_id": key}, null, null, {remove: true}, function(err, res) { + if (err || !res["value"]) + return callback(err); + + if (res["value"]["expire"] > new Date(Date.now() - mongoTimeOffset)) + return callback(null, res["value"]["value"]); + + callback(); + }); +} + + +function mongoLock(lockName, ttl, callback) { + const token = Math.random().toString(36).slice(2); + + function unlockOrExtend(ttl) { + if (!ttl) { + mongoCollection.remove({"_id": lockName, "value": token}, function(err, res) { + if (err || res["result"]["n"] !== 1) + throw err || new Error("Lock expired"); + }); + } + else { + let expire = new Date((Date.now() - mongoTimeOffset) + (ttl * 1000)); + mongoCollection.update( + {"_id": lockName, "value": token}, + {"expire": expire}, + function(err, res) { + if (err || res["result"]["n"] !== 1) + throw err | new Error("Lock expired"); + } + ); + } + } + + let expireTest = new Date(Date.now() - mongoTimeOffset); + let expireSet = new Date((Date.now() - mongoTimeOffset) + (ttl * 1000)); + + mongoCollection.update( + {"_id": lockName, "expire": {"$lte": expireTest}}, + {"$set": {"value" : token, "expire": expireSet}}, + {upsert: true}, + function(err, res) { + if (err && err.code === 11000) + return setTimeout(function() {mongoLock(lockName, ttl, callback)}, 200); + + return callback(err, unlockOrExtend); + } + ); +} + + +exports.connect = connect; +exports.disconnect = disconnect; diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 4234bb7..ef39aee 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -28,6 +28,7 @@ session = require './session' query = require './query' device = require './device' cache = require './cache' +localCache = require './local-cache' db = require './db' logger = require './logger' @@ -213,7 +214,7 @@ appendProvisions = (original, toAppend) -> applyPresets = (sessionContext) -> - cache.getPresets((err, presetsHash, presets) -> + localCache.getPresets((err, presetsHash, presets) -> return throwError(err, sessionContext.httpResponse) if err # Filter presets based on existing faults @@ -437,50 +438,100 @@ nextRpc = (sessionContext) -> ) +endSession = (sessionContext, callback) -> + delete currentSessions.get(sessionContext.httpRequest.connection)[sessionContext.sessionId] + + saveCache = sessionContext.cacheUntil? + + counter = 3 + + counter += 2 + db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + + for k of sessionContext.operationsTouched + counter += 2 + saveCache = true + if sessionContext.operations[k]? + db.saveOperation(sessionContext.deviceId, k, sessionContext.operations[k], (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + else + db.deleteOperation(sessionContext.deviceId, k, (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + + if sessionContext.doneTasks?.length + counter += 2 + saveCache = true + db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + + for k of sessionContext.faultsTouched + counter += 2 + saveCache = true + if sessionContext.faults[k] + sessionContext.faults[k].retries = sessionContext.retries[k] + db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + else + db.deleteFault(sessionContext.deviceId, k, (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + + if saveCache + counter += 2 + cacheDueTasksAndFaultsAndOperations(sessionContext.deviceId, sessionContext.tasks, sessionContext.faults, sessionContext.operations, sessionContext.cacheUntil, (err) -> + if err + callback(err) if counter & 1 + return counter = 0 + + return callback(null, sessionContext.new) if (counter -= 2) == 1 + ) + + callback(null, sessionContext.new) if (counter -= 2) == 1 + + sendAcsRequest = (sessionContext, id, acsRequest) -> if not acsRequest? - session.end(sessionContext, (err, isNew) -> + return endSession(sessionContext, (err, isNew) -> return throwError(err, sessionContext.httpResponse) if err - delete currentSessions.get(sessionContext.httpRequest.connection)[sessionContext.sessionId] if isNew logger.accessInfo({ sessionContext: sessionContext message: 'New device registered' }) - db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) -> - return throwError(err, sessionContext.httpResponse) if err - - counter = 3 - - for k of sessionContext.faultsTouched - counter += 2 - if sessionContext.faults[k] - sessionContext.faults[k].retries = sessionContext.retries[k] - db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) -> - if err - throwError(err, sessionContext.httpResponse) if counter & 1 - return counter = 0 - - if (counter -= 2) == 1 - return writeResponse(sessionContext, soap.response(null)) - ) - else - db.deleteFault(sessionContext.deviceId, k, (err) -> - if err - throwError(err, sessionContext.httpResponse) if counter & 1 - return counter = 0 - - if (counter -= 2) == 1 - return writeResponse(sessionContext, soap.response(null)) - ) - - if (counter -= 2) == 1 - return writeResponse(sessionContext, soap.response(null)) - ) + writeResponse(sessionContext, soap.response(null)) ) - return if acsRequest.name is 'Download' if not acsRequest.url? @@ -493,7 +544,7 @@ sendAcsRequest = (sessionContext, id, acsRequest) -> acsRequest.url += "/#{encodeURIComponent(acsRequest.fileName)}" if not acsRequest.fileSize? - return cache.getFiles((err, hash, files) -> + return localCache.getFiles((err, hash, files) -> return throwError(err, sessionContext.httpResponse) if err if acsRequest.fileName of files @@ -535,7 +586,7 @@ getSession = (httpRequest, callback) -> return callback(null, sessionContext) setTimeout(() -> - db.redisClient.eval('local v=redis.call("get",KEYS[1]);redis.call("del",KEYS[1]);return v;', 1, "session_#{sessionId}", (err, sessionContextString) -> + cache.pop("session_#{sessionId}", (err, sessionContextString) -> return callback(err) if err or not sessionContextString session.deserialize(sessionContextString, (err, sessionContext) -> return callback(err) if err @@ -560,13 +611,75 @@ onConnection = (socket) -> session.serialize(sessionContext, (err, sessionContextString) -> return throwError(err) if err # TODO don't set if process is shutting down - db.redisClient.setex("session_#{sessionId}", sessionContext.timeout, sessionContextString, (err) -> + cache.set("session_#{sessionId}", sessionContextString, sessionContext.timeout, (err) -> return throwError(err) if err ) ) ) +getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) -> + cache.get("#{deviceId}_tasks_faults_operations", (err, res) -> + return callback(err) if err + + if res + res = JSON.parse(res) + return callback(null, res.tasks or [], res.faults or {}, res.operations or {}) + + faults = tasks = operations = cacheUntil = null + + db.getFaults(deviceId, (err, _faults) -> + if err + callback?(err) + return callback = null + + faults = _faults + if tasks? and operations? + return callback(null, tasks, faults, operations, cacheUntil) + ) + + db.getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) -> + if err + callback?(err) + return callback = null + + tasks = dueTasks + cacheUntil = nextTimestamp or 0 + + if faults? and operations? + return callback(null, tasks, faults, operations, cacheUntil) + ) + + db.getOperations(deviceId, (err, _operations) -> + if err + callback?(err) + return callback = null + + operations = _operations + + if faults? and tasks? + return callback(null, tasks, faults, operations, cacheUntil) + ) + + if faults? and tasks? and operations? + return callback(null, tasks, faults, operations, cacheUntil) + ) + + +cacheDueTasksAndFaultsAndOperations = (deviceId, tasks, faults, operations, cacheUntil, callback) -> + v = {} + v.tasks = tasks if tasks.length + v.faults = faults if Object.keys(faults).length + v.operations = operations if Object.keys(operations).length + + if cacheUntil + ttl = Math.trunc((Date.now() - cacheUntil) / 1000) + else + ttl = config.get('MAX_CACHE_TTL', deviceId) + + cache.set("#{deviceId}_tasks_faults_operations", JSON.stringify(v), ttl, callback) + + listener = (httpRequest, httpResponse) -> if httpRequest.method != 'POST' httpResponse.writeHead 405, {'Allow': 'POST'} @@ -626,7 +739,7 @@ listener = (httpRequest, httpResponse) -> httpRequest.connection.setTimeout(sessionContext.timeout * 1000) - db.getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations) -> + getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations, cacheUntil) -> return throwError(err, httpResponse) if err sessionContext.tasks = dueTasks sessionContext.faults = faults @@ -634,6 +747,7 @@ listener = (httpRequest, httpResponse) -> for k, v of faults sessionContext.retries[k] = v.retries sessionContext.operations = operations + sessionContext.cacheUntil = cacheUntil # Delete expired faults for k, v of sessionContext.faults diff --git a/lib/db.coffee b/lib/db.coffee index efaa21c..e70b1e7 100644 --- a/lib/db.coffee +++ b/lib/db.coffee @@ -19,10 +19,8 @@ config = require './config' mongodb = require 'mongodb' -redis = require('redis') common = require './common' -redisClient = null tasksCollection = null devicesCollection = null presetsCollection = null @@ -35,7 +33,7 @@ operationsCollection = null connect = (callback) -> - callbackCounter = 10 + callbackCounter = 9 mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), (err, db) -> return callback(err) if err exports.mongoDb = db @@ -105,18 +103,11 @@ connect = (callback) -> return callback(err) ) - exports.redisClient = redisClient = redis.createClient(config.get('REDIS_PORT'), config.get('REDIS_HOST')) - redisClient.select(config.get('REDIS_DB'), (err) -> - if --callbackCounter == 0 or err - callbackCounter = 0 - return callback(err) - ) ) disconnect = () -> exports.mongoDb?.close() - redisClient?.quit() # Optimize projection by removing overlaps @@ -461,71 +452,6 @@ saveDevice = (deviceId, deviceData, isNew, callback) -> ) -getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) -> - redisClient.mget("#{deviceId}_tasks", "#{deviceId}_faults", "#{deviceId}_operations", (err, res) -> - return callback(err) if err - - if res[0] - tasks = JSON.parse(res[0]) - - if res[1]? - faults = JSON.parse(res[1]) - - if res[2]? - operations = JSON.parse(res[2]) - - if tasks? and faults? and operations? - return callback(null, tasks, faults, operations) - - MAX_CACHE_TTL = config.get('MAX_CACHE_TTL', deviceId) - - if not faults? - getFaults(deviceId, (err, flts) -> - if err - callback?(err) - return callback = null - - faults = flts - if tasks? and operations? - return callback(null, tasks, faults, operations) - ) - - if not tasks? - getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) -> - if err - callback?(err) - return callback = null - - if nextTimestamp? - exp = Math.min(0, Math.trunc((nextTimestamp - Date.now()) / 1000)) - else - exp = MAX_CACHE_TTL - - redisClient.setex("#{deviceId}_tasks", exp, JSON.stringify(dueTasks), (err) -> - if err - callback?(err) - return callback = null - - tasks = dueTasks - if faults? and operations? - return callback(null, tasks, faults, operations) - ) - ) - - if not operations? - getOperations(deviceId, (err, _operations) -> - if err - callback?(err) - return callback = null - - operations = _operations - if tasks? and faults? - return callback(null, tasks, faults, operations) - ) - - ) - - getFaults = (deviceId, callback) -> faultsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) -> return callback(err) if err @@ -540,32 +466,22 @@ getFaults = (deviceId, callback) -> r.provisions = JSON.parse(r.provisions) faults[channel] = r - MAX_CACHE_TTL = config.get('MAX_CACHE_TTL', deviceId) - - redisClient.setex("#{deviceId}_faults", MAX_CACHE_TTL, JSON.stringify(faults), (err) -> - callback(err, faults) - ) + callback(err, faults) ) saveFault = (deviceId, channel, fault, callback) -> + fault = Object.assign({}, fault) fault._id = "#{deviceId}:#{channel}" fault.device = deviceId fault.channel = channel fault.timestamp = new Date(fault.timestamp) fault.provisions = JSON.stringify(fault.provisions) - faultsCollection.save(fault, (err) -> - return callback(err) if err - redisClient.del("#{deviceId}_faults", callback) - ) + faultsCollection.save(fault, callback) deleteFault = (deviceId, channel, callback) -> - faultsCollection.remove({_id: "#{deviceId}:#{channel}"}, (err) -> - return callback(err) if err - redisClient.del("#{deviceId}_faults", callback) - ) - + faultsCollection.remove({_id: "#{deviceId}:#{channel}"}, callback) getDueTasks = (deviceId, timestamp, callback) -> @@ -608,13 +524,7 @@ getDueTasks = (deviceId, timestamp, callback) -> clearTasks = (deviceId, taskIds, callback) -> - if not taskIds?.length - return callback() - - tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, (err) -> - return callback(err) if err - redisClient.del("#{deviceId}_tasks", callback) - ) + tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, callback) getOperations = (deviceId, callback) -> @@ -631,41 +541,58 @@ getOperations = (deviceId, callback) -> r.retries = JSON.parse(r.retries) operations[commandKey] = r - MAX_CACHE_TTL = config.get('MAX_CACHE_TTL', deviceId) - - redisClient.setex("#{deviceId}_operations", MAX_CACHE_TTL, JSON.stringify(operations), (err) -> - callback(err, operations) - ) + callback(err, operations) ) saveOperation = (deviceId, commandKey, operation, callback) -> + operation = Object.assign({}, operation) operation._id = "#{deviceId}:#{commandKey}" operation.timestamp = new Date(operation.timestamp) operation.provisions = JSON.stringify(operation.provisions) operation.retries = JSON.stringify(operation.retries) operation.args = JSON.stringify(operation.args) - operationsCollection.save(operation, (err) -> - return callback(err) if err - redisClient.del("#{deviceId}_operations", callback) - ) + operationsCollection.save(operation, callback) deleteOperation = (deviceId, commandKey, callback) -> - operationsCollection.remove({_id: "#{deviceId}:#{commandKey}"}, (err) -> - return callback(err) if err - redisClient.del("#{deviceId}_operations", callback) - ) + operationsCollection.remove({_id: "#{deviceId}:#{commandKey}"}, callback) + + +getPresets = (callback) -> + presetsCollection.find().toArray(callback) + + +getObjects = (callback) -> + objectsCollection.find().toArray(callback) + + +getProvisions = (callback) -> + provisionsCollection.find().toArray(callback) + + +getVirtualParameters = (callback) -> + virtualParametersCollection.find().toArray(callback) + + +getFiles = (callback) -> + filesCollection.find().toArray(callback) exports.connect = connect exports.disconnect = disconnect exports.fetchDevice = fetchDevice exports.saveDevice = saveDevice -exports.getDueTasksAndFaultsAndOperations = getDueTasksAndFaultsAndOperations exports.getFaults = getFaults exports.saveFault = saveFault exports.deleteFault = deleteFault exports.clearTasks = clearTasks exports.saveOperation = saveOperation exports.deleteOperation = deleteOperation +exports.getPresets = getPresets +exports.getObjects = getObjects +exports.getProvisions = getProvisions +exports.getVirtualParameters = getVirtualParameters +exports.getFiles = getFiles +exports.getDueTasks = getDueTasks +exports.getOperations = getOperations diff --git a/lib/local-cache.js b/lib/local-cache.js new file mode 100644 index 0000000..69d2ef1 --- /dev/null +++ b/lib/local-cache.js @@ -0,0 +1,339 @@ +/** + * Copyright 2013-2017 Zaid Abdulla + * + * This file is part of GenieACS. + * + * GenieACS is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * GenieACS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with GenieACS. If not, see . + */ +"use strict"; + +const vm = require("vm"); +const crypto = require("crypto"); + +const later = require("later"); + +const db = require("./db"); +const cache = require("./cache"); +const query = require("./query"); +const logger = require("./logger"); + +const REFRESH = 3000; + +let nextRefresh = 1; +let hash = null; +let presets, provisions, virtualParameters, files; + + +function computeHash() { + // MD5 hash for presets, provisions, virtual parameters for detecting changes + let h = crypto.createHash("md5"); + for (let p of presets) { + h.update(JSON.stringify(p.name)); + h.update(JSON.stringify(p.channel)); + h.update(JSON.stringify(p.schedule)); + h.update(JSON.stringify(p.events)); + h.update(JSON.stringify(p.precondition)); + h.update(JSON.stringify(p.provisions)); + } + + let keys; + + keys = Object.keys(provisions).sort(); + h.update(JSON.stringify(keys)); + for (let k of keys) + h.update(provisions[k].md5); + + keys = Object.keys(virtualParameters).sort(); + h.update(JSON.stringify(keys)); + for (let k of keys) + h.update(virtualParameters[k].md5); + + hash = h.digest("hex"); +} + + +function refresh(callback) { + if (!nextRefresh) + return setTimeout(function() {refresh(callback);}, 20); + + nextRefresh = 0; + const now = Date.now(); + + cache.get("presets_hash", function(err, res) { + if (err) + return callback(err); + + if (hash && res === hash) { + nextRefresh = now + (REFRESH - (now % REFRESH)); + return callback(); + } + + cache.lock("presets_hash_lock", 3000, function(err, unlockOrExtend) { + if (err) + return callback(err); + + let counter = 3; + + counter += 2; + db.getPresets(function(err, res) { + if (err) { + if (counter & 1) + callback(err); + return counter = 0; + } + + db.getObjects(function(err, objects) { + if (err) { + if (counter & 1) + callback(err); + return counter = 0; + } + + res.sort((a, b) => { + if (a.weight === b.weight) + return a._id > b._id; + else + return a.weight - b.weight; + }); + + presets = []; + for (let preset of res) { + let schedule = null; + if (preset.schedule) { + let parts = preset.schedule.trim().split(/\s+/); + schedule = { + md5: crypto.createHash("md5").update(preset.schedule).digest("hex") + }; + + try { + schedule.duration = +(parts.shift()) * 1000; + if (parts.length === 5) { + parts.unshift("0"); + // TODO later.js doesn't throw erorr if expression is invalid! + schedule.schedule = later.schedule(later.parse.cron(parts.join(" "), true)); + } + } + catch (err) { + logger.warn({ + message: "Invalid preset schedule", + preset: preset._id, + schedule: preset.schedule + }); + schedule.schedule = false; + } + } + + let events = preset.events || {}; + let precondition = query.convertMongoQueryToFilters(JSON.parse(preset.precondition)); + let _provisions = preset.provisions || []; + + // Generate provisions from the old configuration format + for (let c of preset.configurations) { + switch (c.type) { + case "age": + _provisions.push(["refresh", c.name, c.age]); + break; + + case "value": + _provisions.push(["value", c.name, c.value]); + break; + + case "add_tag": + _provisions.push(["tag", c.tag, true]); + break; + + case "delete_tag": + _provisions.push(["tag", c.tag, false]); + break; + + case "provision": + _provisions.push([c.name].concat(c.args || [])); + break; + + case "add_object": + for (let obj of objects) { + if (obj["_id"] === c.object) { + let alias = obj["_keys"].map((k) => `${k}:${JSON.stringify(obj[k])}`).join(","); + let p = `${c.name}.[${alias}]`; + _provisions.push(["instances", p, 1]); + + for (let k in obj) + if (k[0] !== "_" && !(obj["_keys"])) + _provisions.push(["value", `${p}.${k}`, obj[k]]); + } + } + break; + + case "delete_object": + for (let obj of objects) { + if (obj["_id"] === c.object) { + let alias = obj["_keys"].map((k) => `${k}:${JSON.stringify(obj[k])}`).join(","); + let p = `${c.name}.[${alias}]`; + _provisions.push(["instances", p, 0]); + } + } + break; + + default: + if (counter & 1) + callback(new Error(`Unknown configuration type ${c.type}`)); + return counter = 0; + break; + } + } + + presets.push({ + name: preset._id, + channel: preset.channel || "default", + schedule: schedule, + events: events, + precondition: precondition, + provisions: _provisions + }); + } + + if ((counter -= 2) === 1) { + computeHash(); + cache.set("presets_hash", hash, 300, function(err) { + unlockOrExtend(0); + nextRefresh = now + (REFRESH - (now % REFRESH)); + callback(err); + }); + } + }); + }); + + counter += 2; + db.getProvisions(function(err, res) { + if (err) { + if (counter & 1) + callback(err); + return counter = 0; + } + + provisions = {}; + for (let r of res) { + provisions[r._id] = {}; + provisions[r._id].md5 = crypto.createHash("md5").update(r.script).digest("hex"); + provisions[r._id].script = new vm.Script(`"use strict";(function(){\n${r.script}\n})();`, {filename: r._id, lineOffset: -1, timeout: 50}); + } + + if ((counter -= 2) === 1) { + computeHash(); + cache.set("presets_hash", hash, 300, function(err) { + unlockOrExtend(0); + nextRefresh = now + (REFRESH - (now % REFRESH)); + callback(err); + }); + } + }); + + counter += 2; + db.getVirtualParameters(function(err, res) { + if (err) { + if (counter & 1) + callback(err); + return counter = 0; + } + + virtualParameters = {}; + for (let r of res) { + virtualParameters[r._id] = {} + virtualParameters[r._id].md5 = crypto.createHash("md5").update(r.script).digest("hex"); + virtualParameters[r._id].script = new vm.Script(`"use strict";(function(){\n${r.script}\n})();`, {filename: r._id, lineOffset: -1, timeout: 50}); + } + + if ((counter -= 2) === 1) { + computeHash(); + cache.set("presets_hash", hash, 300, function(err) { + unlockOrExtend(0); + nextRefresh = now + (REFRESH - (now % REFRESH)); + callback(err); + }); + } + }); + + counter += 2; + db.getFiles(function(err, res) { + if (err) { + if (counter & 1) + callback(err); + return counter = 0; + } + + files = {}; + for (let r of res) { + const id = r.filename || r._id.toString(); + files[id] = {}; + files[id].length = r.length; + files[id].md5 = r.md5; + files[id].contentType = r.contentType; + } + + if ((counter -= 2) === 1) { + computeHash(); + cache.set("presets_hash", hash, 300, function(err) { + unlockOrExtend(0); + nextRefresh = now + (REFRESH - (now % REFRESH)); + callback(err); + }); + } + }); + + if ((counter -= 2) === 1) { + computeHash(); + cache.set("presets_hash", hash, 300, function(err) { + unlockOrExtend(0); + nextRefresh = now + (REFRESH - (now % REFRESH)); + callback(err); + }); + } + }); + }); +} + + +function getPresets(callback) { + if (Date.now() < nextRefresh) + return callback(null, hash, presets); + + refresh(function(err) { + callback(err, hash, presets); + }); +} + + +function getFiles(callback) { + if (Date.now() < nextRefresh) + return callback(null, hash, files); + + refresh(function(err) { + callback(err, hash, files); + }) +} + + +function getProvisionsAndVirtualParameters(callback) { + if (Date.now() < nextRefresh) + return callback(null, hash, provisions, virtualParameters); + + refresh(function(err) { + callback(err, hash, provisions, virtualParameters); + }); +} + + +exports.getPresets = getPresets; +exports.getFiles = getFiles; +exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters; diff --git a/lib/nbi.coffee b/lib/nbi.coffee index 037741c..95e7566 100644 --- a/lib/nbi.coffee +++ b/lib/nbi.coffee @@ -50,6 +50,7 @@ common = require './common' db = require './db' query = require './query' apiFunctions = require './api-functions' +cache = require './cache' VERSION = require('../package.json').version @@ -108,7 +109,7 @@ listener = (request, response) -> db.presetsCollection.save(preset, (err) -> return throwError(err, response) if err - db.redisClient.del('presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -117,7 +118,7 @@ listener = (request, response) -> else if request.method == 'DELETE' db.presetsCollection.remove({'_id' : presetName}, (err) -> return throwError(err, response) if err - db.redisClient.del('presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -134,7 +135,7 @@ listener = (request, response) -> db.objectsCollection.save(object, (err) -> return throwError(err, response) if err - db.redisClient.del('objects', 'presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -143,7 +144,7 @@ listener = (request, response) -> else if request.method == 'DELETE' db.objectsCollection.remove({'_id' : objectName}, (err) -> return throwError(err, response) if err - db.redisClient.del('objects', 'presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -168,7 +169,7 @@ listener = (request, response) -> db.provisionsCollection.save(object, (err) -> return throwError(err, response) if err - db.redisClient.del('presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -177,7 +178,7 @@ listener = (request, response) -> else if request.method == 'DELETE' db.provisionsCollection.remove({'_id' : provisionName}, (err) -> return throwError(err, response) if err - db.redisClient.del('presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -202,7 +203,7 @@ listener = (request, response) -> db.virtualParametersCollection.save(object, (err) -> return throwError(err, response) if err - db.redisClient.del('presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -211,7 +212,7 @@ listener = (request, response) -> else if request.method == 'DELETE' db.virtualParametersCollection.remove({'_id' : virtualParameterName}, (err) -> return throwError(err, response) if err - db.redisClient.del('presets_hash', (err) -> + cache.del('presets_hash', (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -227,20 +228,14 @@ listener = (request, response) -> if request.method == 'POST' db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_presets_hash", (err) -> - return throwError(err, response) if err - response.writeHead(200) - response.end() - ) + response.writeHead(200) + response.end() ) else if request.method == 'DELETE' db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_presets_hash", (err) -> - return throwError(err, response) if err - response.writeHead(200) - response.end() - ) + response.writeHead(200) + response.end() ) else response.writeHead 405, {'Allow': 'POST, DELETE'} @@ -255,14 +250,14 @@ listener = (request, response) -> if channel.startsWith('task_') return db.tasksCollection.remove({_id : new mongodb.ObjectID(channel.slice(5))}, (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) -> + cache.del("#{deviceId}_tasks_faults_operations", (err) -> return throwError(err, response) if err response.writeHead(200) response.end() ) ) - db.redisClient.del("#{deviceId}_faults", (err) -> + cache.del("#{deviceId}_tasks_faults_operations", (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -279,7 +274,7 @@ listener = (request, response) -> task.device = deviceId apiFunctions.insertTasks(task, (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_tasks", (err) -> + cache.del("#{deviceId}_tasks_faults_operations", (err) -> return throwError(err, response) if err if urlParts.query.connection_request? @@ -340,7 +335,7 @@ listener = (request, response) -> return throwError(err, response) if err db.faultsCollection.remove({_id : "#{deviceId}:task_#{String(taskId)}"}, (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) -> + cache.del("#{deviceId}_tasks_faults_operations", (err) -> return throwError(err, response) if err response.writeHead(200) response.end() @@ -358,7 +353,7 @@ listener = (request, response) -> deviceId = task.device db.faultsCollection.remove({_id : "#{deviceId}:task_#{String(taskId)}"}, (err) -> return throwError(err, response) if err - db.redisClient.del("#{deviceId}_faults", (err) -> + cache.del("#{deviceId}_tasks_faults_operations", (err) -> return throwError(err, response) if err response.writeHead(200) response.end() diff --git a/lib/server.coffee b/lib/server.coffee index 84f6fdc..7398039 100644 --- a/lib/server.coffee +++ b/lib/server.coffee @@ -27,6 +27,7 @@ logger = require './logger' logger.init(service, require('../package.json').version) config = require './config' db = require './db' +cache = require './cache' extensions = require './extensions' networkInterface = config.get("#{service.toUpperCase()}_INTERFACE") @@ -47,12 +48,14 @@ exit = () -> if not server db.disconnect() + cache.disconnect() extensions.killAll() logger.close() return server.close(() -> db.disconnect() + cache.disconnect() extensions.killAll() logger.close() ) @@ -91,7 +94,10 @@ else db.connect((err) -> throw err if err - server.listen(port, networkInterface) + cache.connect((err) -> + throw err if err + server.listen(port, networkInterface) + ) ) diff --git a/lib/session.coffee b/lib/session.coffee index 2b6ff05..6c09756 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -22,7 +22,7 @@ common = require './common' db = require './db' device = require './device' sandbox = require './sandbox' -cache = require './cache' +localCache = require './local-cache' PathSet = require './path-set' VersionedMap = require './versioned-map' InstanceSet = require './instance-set' @@ -65,7 +65,7 @@ init = (deviceId, cwmpVersion, timeout, callback) -> declarations: [] } - cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> + localCache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> return callback(err) if err sessionContext.presetsHash = hash @@ -1515,7 +1515,7 @@ rpcFault = (sessionContext, id, faultResponse, callback) -> deserialize = (sessionContextString, callback) -> - cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> + localCache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> return callback(err) if err sessionContext = JSON.parse(sessionContextString) @@ -1581,48 +1581,6 @@ serialize = (sessionContext, callback) -> return callback(null, sessionContextString) -end = (sessionContext, callback) -> - counter = 3 - - counter += 2 - db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, (err) -> - if err - callback(err) if counter & 1 - return counter = 0 - - return callback(null, sessionContext.new) if (counter -= 2) == 1 - ) - - counter += 2 - db.redisClient.del("session_#{sessionContext.id}", (err) -> - if err - callback(err) if counter & 1 - return counter = 0 - - return callback(null, sessionContext.new) if (counter -= 2) == 1 - ) - - for k of sessionContext.operationsTouched - counter += 2 - if sessionContext.operations[k]? - db.saveOperation(sessionContext.deviceId, k, sessionContext.operations[k], (err) -> - if err - callback(err) if counter & 1 - return counter = 0 - - return callback(null, sessionContext.new) if (counter -= 2) == 1 - ) - else - db.deleteOperation(sessionContext.deviceId, k, (err) -> - if err - callback(err) if counter & 1 - return counter = 0 - - return callback(null, sessionContext.new) if (counter -= 2) == 1 - ) - - callback(null, sessionContext.new) if (counter -= 2) == 1 - exports.init = init exports.timeoutOperations = timeoutOperations exports.inform = inform @@ -1632,6 +1590,5 @@ exports.clearProvisions = clearProvisions exports.rpcRequest = rpcRequest exports.rpcResponse = rpcResponse exports.rpcFault = rpcFault -exports.end = end exports.serialize = serialize exports.deserialize = deserialize