From e909d783e239b2d953ee22bc8c18650d9d62f249 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Fri, 14 Dec 2018 13:24:59 -0800 Subject: [PATCH] Refactor local cache --- lib/cwmp.js | 341 +++++++++++++++---------------- lib/local-cache.js | 491 +++++++++++++++++++++++---------------------- lib/session.js | 77 +++---- 3 files changed, 444 insertions(+), 465 deletions(-) diff --git a/lib/cwmp.js b/lib/cwmp.js index 091e9ae..3842735 100644 --- a/lib/cwmp.js +++ b/lib/cwmp.js @@ -278,203 +278,195 @@ function appendProvisions(original, toAppend) { function applyPresets(sessionContext) { const deviceData = sessionContext.deviceData; - localCache.getPresets((err, presetsHash, presets) => { - if (err) return void throwError(err, sessionContext.httpResponse); + const presets = localCache.getPresets(sessionContext.cacheSnapshot); - // Filter presets based on existing faults - const blackList = {}; - let whiteList = null; - let whiteListProvisions = null; - const RETRY_DELAY = config.get("RETRY_DELAY", sessionContext.deviceId); + // Filter presets based on existing faults + const blackList = {}; + let whiteList = null; + let whiteListProvisions = null; + const RETRY_DELAY = config.get("RETRY_DELAY", sessionContext.deviceId); - if (sessionContext.faults) { - for (const [channel, fault] of Object.entries(sessionContext.faults)) { - let retryTimestamp = 0; - if (!fault.retryNow) { - retryTimestamp = - fault.timestamp + - RETRY_DELAY * Math.pow(2, sessionContext.retries[channel]) * 1000; - } + if (sessionContext.faults) { + for (const [channel, fault] of Object.entries(sessionContext.faults)) { + let retryTimestamp = 0; + if (!fault.retryNow) { + retryTimestamp = + fault.timestamp + + RETRY_DELAY * Math.pow(2, sessionContext.retries[channel]) * 1000; + } - if (retryTimestamp <= sessionContext.timestamp) { - whiteList = channel; - whiteListProvisions = fault.provisions; - break; - } + if (retryTimestamp <= sessionContext.timestamp) { + whiteList = channel; + whiteListProvisions = fault.provisions; + break; + } - blackList[channel] = fault.precondition ? 1 : 2; + blackList[channel] = fault.precondition ? 1 : 2; + } + } + + deviceData.timestamps.revision = 1; + deviceData.attributes.revision = 1; + + const deviceEvents = {}; + for (const p of deviceData.paths.find(["Events", "*"], false, true)) { + const attrs = deviceData.attributes.get(p); + if (attrs && attrs.value && attrs.value[1][0] >= sessionContext.timestamp) + deviceEvents[p[1]] = true; + } + + const parameters = {}; + const filteredPresets = []; + + for (const preset of presets) { + if (whiteList != null) { + if (preset.channel !== whiteList) continue; + } else if (blackList[preset.channel] === 1) { + continue; + } + + let eventsMatch = true; + for (const [k, v] of Object.entries(preset.events)) { + if (!v !== !deviceEvents[k.replace(/\s+/g, "_")]) { + eventsMatch = false; + break; } } - deviceData.timestamps.revision = 1; - deviceData.attributes.revision = 1; + if (!eventsMatch) continue; - const deviceEvents = {}; - for (const p of deviceData.paths.find(["Events", "*"], false, true)) { - const attrs = deviceData.attributes.get(p); - if (attrs && attrs.value && attrs.value[1][0] >= sessionContext.timestamp) - deviceEvents[p[1]] = true; - } - - const parameters = {}; - const filteredPresets = []; - - for (const preset of presets) { - if (whiteList != null) { - if (preset.channel !== whiteList) continue; - } else if (blackList[preset.channel] === 1) { + if (preset.schedule && preset.schedule.schedule) { + const r = scheduling.cron( + sessionContext.timestamp, + preset.schedule.schedule + ); + if (!(r[0] + preset.schedule.duration > sessionContext.timestamp)) continue; + } + + filteredPresets.push(preset); + for (const k of Object.keys(preset.precondition)) { + sessionContext.channels[preset.channel] = 0; + const p = k.split(/([^a-zA-Z0-9\-_.].*)/, 1)[0]; + parameters[p] = common.parsePath(p); + } + } + + const declarations = Object.values(parameters).map(v => ({ + path: v, + pathGet: 1, + pathSet: null, + attrGet: { value: 1 }, + attrSet: null, + defer: true + })); + + session.rpcRequest( + sessionContext, + declarations, + (err, flt, reqId, acsReq) => { + if (err) return void throwError(err, sessionContext.httpResponse); + + if (flt) { + recordFault(sessionContext, flt); + session.clearProvisions(sessionContext); + return void applyPresets(sessionContext); } - let eventsMatch = true; - for (const [k, v] of Object.entries(preset.events)) { - if (!v !== !deviceEvents[k.replace(/\s+/g, "_")]) { - eventsMatch = false; - break; + if (acsReq) return void sendAcsRequest(sessionContext, reqId, acsReq); + + session.clearProvisions(sessionContext); + + const parameterValues = {}; + for (const [k, v] of Object.entries(parameters)) { + const unpacked = device.unpack(deviceData, v); + if (!unpacked[0]) continue; + const attrs = deviceData.attributes.get(unpacked[0]); + if (attrs && attrs.value && attrs.value[1]) + parameterValues[k] = attrs.value[1][0]; + } + + if (whiteList != null) + session.addProvisions(sessionContext, whiteList, whiteListProvisions); + + const appendProvisionsToFaults = {}; + for (const p of filteredPresets) { + if (query.testFilter(parameterValues, p.precondition)) { + if (blackList[p.channel] === 2) { + appendProvisionsToFaults[p.channel] = ( + appendProvisionsToFaults[p.channel] || [] + ).concat(p.provisions); + } else { + session.addProvisions(sessionContext, p.channel, p.provisions); + } } } - if (!eventsMatch) continue; - - if (preset.schedule && preset.schedule.schedule) { - const r = scheduling.cron( - sessionContext.timestamp, - preset.schedule.schedule - ); - if (!(r[0] + preset.schedule.duration > sessionContext.timestamp)) - continue; + for (const [channel, provisions] of Object.entries( + appendProvisionsToFaults + )) { + if ( + appendProvisions( + sessionContext.faults[channel].provisions, + provisions + ) + ) { + if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {}; + sessionContext.faultsTouched[channel] = true; + } } - filteredPresets.push(preset); - for (const k of Object.keys(preset.precondition)) { - sessionContext.channels[preset.channel] = 0; - const p = k.split(/([^a-zA-Z0-9\-_.].*)/, 1)[0]; - parameters[p] = common.parsePath(p); + // Don't increment when processing a single channel (e.g. after fault) + if (whiteList == null) + sessionContext.presetCycles = (sessionContext.presetCycles || 0) + 1; + + if (sessionContext.presetCycles > MAX_CYCLES) { + const fault = { + code: "preset_loop", + message: "The presets are stuck in an endless configuration loop", + timestamp: sessionContext.timestamp + }; + recordFault(sessionContext, fault); + // No need to save retryNow + for (const f of Object.values(sessionContext.faults)) delete f.retryNow; + session.clearProvisions(sessionContext); + return void sendAcsRequest(sessionContext); } - } - const declarations = Object.values(parameters).map(v => ({ - path: v, - pathGet: 1, - pathSet: null, - attrGet: { value: 1 }, - attrSet: null, - defer: true - })); - - session.rpcRequest( - sessionContext, - declarations, - (err, flt, reqId, acsReq) => { + deviceData.timestamps.dirty = 0; + deviceData.attributes.dirty = 0; + session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) => { if (err) return void throwError(err, sessionContext.httpResponse); - if (flt) { - recordFault(sessionContext, flt); + if (fault) { + recordFault(sessionContext, fault); session.clearProvisions(sessionContext); return void applyPresets(sessionContext); } - if (acsReq) return void sendAcsRequest(sessionContext, reqId, acsReq); - - session.clearProvisions(sessionContext); - - const parameterValues = {}; - for (const [k, v] of Object.entries(parameters)) { - const unpacked = device.unpack(deviceData, v); - if (!unpacked[0]) continue; - const attrs = deviceData.attributes.get(unpacked[0]); - if (attrs && attrs.value && attrs.value[1]) - parameterValues[k] = attrs.value[1][0]; - } - - if (whiteList != null) - session.addProvisions(sessionContext, whiteList, whiteListProvisions); - - const appendProvisionsToFaults = {}; - for (const p of filteredPresets) { - if (query.testFilter(parameterValues, p.precondition)) { - if (blackList[p.channel] === 2) { - appendProvisionsToFaults[p.channel] = ( - appendProvisionsToFaults[p.channel] || [] - ).concat(p.provisions); - } else { - session.addProvisions(sessionContext, p.channel, p.provisions); + if (!acsRequest) { + for (const channel of Object.keys(sessionContext.channels)) { + if (sessionContext.faults[channel]) { + delete sessionContext.faults[channel]; + if (!sessionContext.faultsTouched) + sessionContext.faultsTouched = {}; + sessionContext.faultsTouched[channel] = true; } } - } - for (const [channel, provisions] of Object.entries( - appendProvisionsToFaults - )) { + if (whiteList != null) return void applyPresets(sessionContext); + if ( - appendProvisions( - sessionContext.faults[channel].provisions, - provisions - ) - ) { - if (!sessionContext.faultsTouched) - sessionContext.faultsTouched = {}; - sessionContext.faultsTouched[channel] = true; - } + sessionContext.deviceData.timestamps.dirty > 1 || + sessionContext.deviceData.attributes.dirty > 1 + ) + return void applyPresets(sessionContext); } - // Don't increment when processing a single channel (e.g. after fault) - if (whiteList == null) - sessionContext.presetCycles = (sessionContext.presetCycles || 0) + 1; - - if (sessionContext.presetCycles > MAX_CYCLES) { - const fault = { - code: "preset_loop", - message: "The presets are stuck in an endless configuration loop", - timestamp: sessionContext.timestamp - }; - recordFault(sessionContext, fault); - // No need to save retryNow - for (const f of Object.values(sessionContext.faults)) - delete f.retryNow; - session.clearProvisions(sessionContext); - return void sendAcsRequest(sessionContext); - } - - deviceData.timestamps.dirty = 0; - deviceData.attributes.dirty = 0; - session.rpcRequest( - sessionContext, - null, - (err, fault, id, acsRequest) => { - if (err) return void throwError(err, sessionContext.httpResponse); - - if (fault) { - recordFault(sessionContext, fault); - session.clearProvisions(sessionContext); - return void applyPresets(sessionContext); - } - - if (!acsRequest) { - for (const channel of Object.keys(sessionContext.channels)) { - if (sessionContext.faults[channel]) { - delete sessionContext.faults[channel]; - if (!sessionContext.faultsTouched) - sessionContext.faultsTouched = {}; - sessionContext.faultsTouched[channel] = true; - } - } - - if (whiteList != null) return void applyPresets(sessionContext); - - if ( - sessionContext.deviceData.timestamps.dirty > 1 || - sessionContext.deviceData.attributes.dirty > 1 - ) - return void applyPresets(sessionContext); - } - - sendAcsRequest(sessionContext, id, acsRequest); - } - ); - } - ); - }); + sendAcsRequest(sessionContext, id, acsRequest); + }); + } + ); } function nextRpc(sessionContext) { @@ -718,6 +710,7 @@ function sendAcsRequest(sessionContext, id, acsRequest) { return void writeResponse(sessionContext, soap.response(null), true); if (acsRequest.name === "Download") { + acsRequest.fileSize = 0; if (!acsRequest.url) { const FS_PORT = config.get("FS_PORT"); const FS_HOSTNAME = config.get("FS_HOSTNAME"); @@ -726,18 +719,10 @@ function sendAcsRequest(sessionContext, id, acsRequest) { acsRequest.url += FS_HOSTNAME; if (FS_PORT !== 80) acsRequest.url += ":" + FS_PORT; acsRequest.url += "/" + encodeURI(acsRequest.fileName); - } - if (acsRequest.fileSize == null) { - return void localCache.getFiles((err, hash, files) => { - if (err) return void throwError(err, sessionContext.httpResponse); - - if (files[acsRequest.fileName]) - acsRequest.fileSize = files[acsRequest.fileName].length; - else acsRequest.fileSize = 0; - - sendAcsRequest(sessionContext, id, acsRequest); - }); + const files = localCache.getFiles(sessionContext.cacheSnapshot); + if (files[acsRequest.fileName]) + acsRequest.fileSize = files[acsRequest.fileName].length; } } diff --git a/lib/local-cache.js b/lib/local-cache.js index bc241a3..078c7a5 100644 --- a/lib/local-cache.js +++ b/lib/local-cache.js @@ -28,15 +28,16 @@ const logger = require("./logger"); const scheduling = require("./scheduling"); const REFRESH = 3000; +const EVICT_TIMEOUT = 60000; +const snapshots = new Map(); +let currentSnapshot = null; let nextRefresh = 1; -let hash = null; -let presets, provisions, virtualParameters, files; -function computeHash() { +function computeHash(snapshot) { // MD5 hash for presets, provisions, virtual parameters for detecting changes const h = crypto.createHash("md5"); - for (const p of presets) { + for (const p of snapshot.presets) { h.update(JSON.stringify(p.name)); h.update(JSON.stringify(p.channel)); h.update(JSON.stringify(p.schedule)); @@ -47,15 +48,15 @@ function computeHash() { let keys; - keys = Object.keys(provisions).sort(); + keys = Object.keys(snapshot.provisions).sort(); h.update(JSON.stringify(keys)); - for (const k of keys) h.update(provisions[k].md5); + for (const k of keys) h.update(snapshot.provisions[k].md5); - keys = Object.keys(virtualParameters).sort(); + keys = Object.keys(snapshot.virtualParameters).sort(); h.update(JSON.stringify(keys)); - for (const k of keys) h.update(virtualParameters[k].md5); + for (const k of keys) h.update(snapshot.virtualParameters[k].md5); - hash = h.digest("hex"); + return h.digest("hex"); } function flattenObject(src, prefix = "", dst = {}) { @@ -81,7 +82,7 @@ function refresh(callback) { cache.get("presets_hash", (err, dbHash) => { if (err) return void callback(err); - if (hash && dbHash === hash) { + if (currentSnapshot && dbHash === currentSnapshot) { nextRefresh = now + (REFRESH - (now % REFRESH)); return void callback(); } @@ -89,277 +90,281 @@ function refresh(callback) { cache.lock("presets_hash_lock", 3, (err, unlockOrExtend) => { if (err) return void callback(err); - let counter = 3; + const promises = []; - counter += 2; - db.getPresets((err, res) => { - if (err) { - if (counter & 1) callback(err); - return void (counter = 0); - } + promises.push( + new Promise((resolve, reject) => { + db.getPresets((err, res) => { + if (err) return void reject(err); - db.getObjects((err, objects) => { - if (err) { - if (counter & 1) callback(err); - return void (counter = 0); - } + db.getObjects((err, objects) => { + if (err) return void reject(err); - objects = objects.map(obj => { - // Flatten object - obj = flattenObject(obj); + objects = objects.map(obj => { + // Flatten object + obj = flattenObject(obj); - // If no keys are defined, consider all parameters as keys to keep the - // same behavior from v1.0 - if (!obj["_keys"] || !obj["_keys"].length) - obj["_keys"] = Object.keys(obj).filter(k => !k.startsWith("_")); + // If no keys are defined, consider all parameters as keys to keep the + // same behavior from v1.0 + if (!obj["_keys"] || !obj["_keys"].length) { + obj["_keys"] = Object.keys(obj).filter( + k => !k.startsWith("_") + ); + } - return obj; - }); + return obj; + }); - res.sort((a, b) => { - if (a.weight === b.weight) return a._id > b._id; - else return a.weight - b.weight; - }); + res.sort((a, b) => { + if (a.weight === b.weight) return a._id > b._id; + else return a.weight - b.weight; + }); - presets = []; - for (const preset of res) { - let schedule = null; - if (preset.schedule) { - const parts = preset.schedule.trim().split(/\s+/); - schedule = { - md5: crypto - .createHash("md5") - .update(preset.schedule) - .digest("hex") - }; + const presets = []; + for (const preset of res) { + let schedule = null; + if (preset.schedule) { + const parts = preset.schedule.trim().split(/\s+/); + schedule = { + md5: crypto + .createHash("md5") + .update(preset.schedule) + .digest("hex") + }; - try { - schedule.duration = +parts.shift() * 1000; - schedule.schedule = scheduling.parseCron(parts.join(" ")); - } catch (err) { - logger.warn({ - message: "Invalid preset schedule", - preset: preset._id, - schedule: preset.schedule - }); - schedule.schedule = false; - } - } + try { + schedule.duration = +parts.shift() * 1000; + schedule.schedule = scheduling.parseCron(parts.join(" ")); + } catch (err) { + logger.warn({ + message: "Invalid preset schedule", + preset: preset._id, + schedule: preset.schedule + }); + schedule.schedule = false; + } + } - const events = preset.events || {}; - const precondition = query.convertMongoQueryToFilters( - JSON.parse(preset.precondition) - ); - const _provisions = preset.provisions || []; + const events = preset.events || {}; + const precondition = query.convertMongoQueryToFilters( + JSON.parse(preset.precondition) + ); + const _provisions = preset.provisions || []; - // Generate provisions from the old configuration format - for (const c of preset.configurations) { - switch (c.type) { - case "age": - _provisions.push(["refresh", c.name, c.age]); - break; + // Generate provisions from the old configuration format + for (const 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 "value": + _provisions.push(["value", c.name, c.value]); + break; - case "add_tag": - _provisions.push(["tag", c.tag, true]); - break; + case "add_tag": + _provisions.push(["tag", c.tag, true]); + break; - case "delete_tag": - _provisions.push(["tag", c.tag, false]); - break; + case "delete_tag": + _provisions.push(["tag", c.tag, false]); + break; - case "provision": - _provisions.push([c.name].concat(c.args || [])); - break; + case "provision": + _provisions.push([c.name].concat(c.args || [])); + break; - case "add_object": - for (const obj of objects) { - if (obj["_id"] === c.object) { - const alias = obj["_keys"] - .map(k => `${k}:${JSON.stringify(obj[k])}`) - .join(","); - const p = `${c.name}.[${alias}]`; - _provisions.push(["instances", p, 1]); + case "add_object": + for (const obj of objects) { + if (obj["_id"] === c.object) { + const alias = obj["_keys"] + .map(k => `${k}:${JSON.stringify(obj[k])}`) + .join(","); + const p = `${c.name}.[${alias}]`; + _provisions.push(["instances", p, 1]); - for (const k in obj) { - if ( - !k.startsWith("_") && - !(obj["_keys"].indexOf(k) !== -1) - ) - _provisions.push(["value", `${p}.${k}`, obj[k]]); + for (const k in obj) { + if ( + !k.startsWith("_") && + !(obj["_keys"].indexOf(k) !== -1) + ) + _provisions.push(["value", `${p}.${k}`, obj[k]]); + } + } } - } + + break; + + case "delete_object": + for (const obj of objects) { + if (obj["_id"] === c.object) { + const alias = obj["_keys"] + .map(k => `${k}:${JSON.stringify(obj[k])}`) + .join(","); + const p = `${c.name}.[${alias}]`; + _provisions.push(["instances", p, 0]); + } + } + + break; + + default: + return void reject( + new Error(`Unknown configuration type ${c.type}`) + ); } + } - break; - - case "delete_object": - for (const obj of objects) { - if (obj["_id"] === c.object) { - const alias = obj["_keys"] - .map(k => `${k}:${JSON.stringify(obj[k])}`) - .join(","); - const p = `${c.name}.[${alias}]`; - _provisions.push(["instances", p, 0]); - } - } - - break; - - default: - if (counter & 1) - callback(new Error(`Unknown configuration type ${c.type}`)); - return void (counter = 0); + presets.push({ + name: preset._id, + channel: preset.channel || "default", + schedule: schedule, + events: events, + precondition: precondition, + provisions: _provisions + }); } + + resolve(presets); + }); + }); + }) + ); + + promises.push( + new Promise((resolve, reject) => { + db.getProvisions((err, res) => { + if (err) return void reject(err); + + const provisions = {}; + for (const 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 } + ); } - presets.push({ - name: preset._id, - channel: preset.channel || "default", - schedule: schedule, - events: events, - precondition: precondition, - provisions: _provisions - }); + resolve(provisions); + }); + }) + ); + + promises.push( + new Promise((resolve, reject) => { + db.getVirtualParameters((err, res) => { + if (err) return void reject(err); + + const virtualParameters = {}; + for (const 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 } + ); + } + + resolve(virtualParameters); + }); + }) + ); + + promises.push( + new Promise((resolve, reject) => { + db.getFiles((err, res) => { + if (err) return void reject(err); + + const files = {}; + for (const 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; + } + + resolve(files); + }); + }) + ); + + Promise.all(promises) + .then(res => { + const snapshot = { + presets: res[0], + provisions: res[1], + virtualParameters: res[2], + files: res[3] + }; + + if (currentSnapshot) { + const h = currentSnapshot; + const s = snapshots.get(h); + setTimeout(() => { + if (snapshots.get(h) === s) snapshots.delete(h); + }, EVICT_TIMEOUT).unref(); } - if ((counter -= 2) === 1) { - computeHash(); - cache.set("presets_hash", hash, 300, err => { - unlockOrExtend(0); - nextRefresh = now + (REFRESH - (now % REFRESH)); - callback(err); - }); - } - }); - }); - - counter += 2; - db.getProvisions((err, res) => { - if (err) { - if (counter & 1) callback(err); - return void (counter = 0); - } - - provisions = {}; - for (const 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, err => { + currentSnapshot = computeHash(snapshot); + snapshots.set(currentSnapshot, snapshot); + cache.set("presets_hash", currentSnapshot, 300, err => { unlockOrExtend(0); + if (err) return void callback(err); nextRefresh = now + (REFRESH - (now % REFRESH)); - callback(err); + callback(); }); - } - }); - - counter += 2; - db.getVirtualParameters((err, res) => { - if (err) { - if (counter & 1) callback(err); - return void (counter = 0); - } - - virtualParameters = {}; - for (const 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, err => { - unlockOrExtend(0); - nextRefresh = now + (REFRESH - (now % REFRESH)); - callback(err); - }); - } - }); - - counter += 2; - db.getFiles((err, res) => { - if (err) { - if (counter & 1) callback(err); - return void (counter = 0); - } - - files = {}; - for (const 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, err => { - unlockOrExtend(0); - nextRefresh = now + (REFRESH - (now % REFRESH)); - callback(err); - }); - } - }); - - if ((counter -= 2) === 1) { - computeHash(); - cache.set("presets_hash", hash, 300, err => { - unlockOrExtend(0); - nextRefresh = now + (REFRESH - (now % REFRESH)); - callback(err); - }); - } + }) + .catch(callback); }); }); } -function getPresets(callback) { - if (Date.now() < nextRefresh) return void callback(null, hash, presets); - +function getCurrentSnapshot(callback) { + if (Date.now() < nextRefresh) return void callback(null, currentSnapshot); refresh(err => { - callback(err, hash, presets); + callback(err, currentSnapshot); }); } -function getFiles(callback) { - if (Date.now() < nextRefresh) return void callback(null, hash, files); - - refresh(err => { - callback(err, hash, files); - }); +function hasSnapshot(hash) { + return snapshots.has(hash); } -function getProvisionsAndVirtualParameters(callback) { - if (Date.now() < nextRefresh) - return void callback(null, hash, provisions, virtualParameters); - - refresh(err => { - callback(err, hash, provisions, virtualParameters); - }); +function getPresets(snapshotKey) { + const snapshot = snapshots.get(snapshotKey); + if (!snapshot) throw new Error("Cache snapshot does not exist"); + return snapshot.presets; } +function getProvisions(snapshotKey) { + const snapshot = snapshots.get(snapshotKey); + if (!snapshot) throw new Error("Cache snapshot does not exist"); + return snapshot.provisions; +} + +function getVirtualParameters(snapshotKey) { + const snapshot = snapshots.get(snapshotKey); + if (!snapshot) throw new Error("Cache snapshot does not exist"); + return snapshot.virtualParameters; +} + +function getFiles(snapshotKey) { + const snapshot = snapshots.get(snapshotKey); + if (!snapshot) throw new Error("Cache snapshot does not exist"); + return snapshot.files; +} + +exports.getCurrentSnapshot = getCurrentSnapshot; +exports.hasSnapshot = hasSnapshot; exports.getPresets = getPresets; +exports.getProvisions = getProvisions; +exports.getVirtualParameters = getVirtualParameters; exports.getFiles = getFiles; -exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters; diff --git a/lib/session.js b/lib/session.js index b3918e4..af2c00a 100644 --- a/lib/session.js +++ b/lib/session.js @@ -34,9 +34,6 @@ const gpnHeuristic = require("./gpn-heuristic"); // phases: read and update const MAX_ITERATIONS = config.get("MAX_COMMIT_ITERATIONS") * 2; -const provisionsCache = new WeakMap(); -const virtualParametersCache = new WeakMap(); - function initDeviceData() { return { paths: new PathSet(), @@ -67,16 +64,11 @@ function init(deviceId, cwmpVersion, timeout, callback) { declarations: [] }; - localCache.getProvisionsAndVirtualParameters( - (err, hash, provisions, virtualParameters) => { - if (err) return void callback(err); - - sessionContext.presetsHash = hash; - provisionsCache.set(sessionContext, provisions); - virtualParametersCache.set(sessionContext, virtualParameters); - callback(null, sessionContext); - } - ); + localCache.getCurrentSnapshot((err, cacheSnapshot) => { + if (err) return void callback(err); + sessionContext.cacheSnapshot = cacheSnapshot; + callback(null, sessionContext); + }); } function loadParameters(sessionContext, callback) { @@ -550,7 +542,7 @@ function runProvisions( let allDeclarations = []; let allClear = []; let counter = 3; - const allProvisions = provisionsCache.get(sessionContext); + const allProvisions = localCache.getProvisions(sessionContext.cacheSnapshot); for (const [j, provision] of provisions.entries()) { if (!allProvisions[provision[0]]) { @@ -616,7 +608,9 @@ function runVirtualParameters( let allDeclarations = []; let allClear = []; let counter = 3; - const allVirtualParameters = virtualParametersCache.get(sessionContext); + const allVirtualParameters = localCache.getVirtualParameters( + sessionContext.cacheSnapshot + ); for (const [j, provision] of provisions.entries()) { counter += 2; @@ -777,7 +771,9 @@ function runDeclarations(sessionContext, declarations) { const allDeclareAttributeTimestamps = new Map(); const allDeclareAttributeValues = new Map(); - const allVirtualParameters = virtualParametersCache.get(sessionContext); + const allVirtualParameters = localCache.getVirtualParameters( + sessionContext.cacheSnapshot + ); function mergeAttributeTimestamps(p, attrs) { let cur = allDeclareAttributeTimestamps.get(p); @@ -1118,7 +1114,9 @@ function rpcRequest(sessionContext, _declarations, callback) { let toClear; - const allVirtualParameters = virtualParametersCache.get(sessionContext); + const allVirtualParameters = localCache.getVirtualParameters( + sessionContext.cacheSnapshot + ); vpd = vpd.filter(declaration => { if (Object.keys(allVirtualParameters).length) { @@ -2595,39 +2593,30 @@ function rpcFault(sessionContext, id, faultResponse, callback) { } function deserialize(sessionContextString, callback) { - localCache.getProvisionsAndVirtualParameters( - (err, hash, provisions, virtualParameters) => { - if (err) return void callback(err); + const sessionContext = JSON.parse(sessionContextString); - const sessionContext = JSON.parse(sessionContextString); + for (const decs of sessionContext.declarations) + for (const d of decs) common.addPathMeta(d.path); - if (sessionContext.presetsHash && sessionContext.presetsHash !== hash) - return void callback(new Error("Preset hash mismatch")); + const deviceData = initDeviceData(); + for (const r of sessionContext.deviceData) { + const path = deviceData.paths.add(r[0]); - provisionsCache.set(sessionContext, provisions); - virtualParametersCache.set(sessionContext, virtualParameters); + if (r[1]) deviceData.loaded.set(path, r[1]); - for (const decs of sessionContext.declarations) - for (const d of decs) common.addPathMeta(d.path); + if (r[2]) deviceData.trackers.set(path, r[2]); - const deviceData = initDeviceData(); - for (const r of sessionContext.deviceData) { - const path = deviceData.paths.add(r[0]); - - if (r[1]) deviceData.loaded.set(path, r[1]); - - if (r[2]) deviceData.trackers.set(path, r[2]); - - if (r[3]) { - deviceData.timestamps.setRevisions(path, r[3]); - if (r[4]) deviceData.attributes.setRevisions(path, r[4]); - } - } - - sessionContext.deviceData = deviceData; - callback(null, sessionContext); + if (r[3]) { + deviceData.timestamps.setRevisions(path, r[3]); + if (r[4]) deviceData.attributes.setRevisions(path, r[4]); } - ); + } + + sessionContext.deviceData = deviceData; + // Ensure cache is populated + localCache.getCurrentSnapshot(err => { + callback(err, sessionContext); + }); } function serialize(sessionContext, callback) {