Refactor local cache
This commit is contained in:
1 parent
8b18f97d12
commit
e909d783e2
3 files changed
+444
-465
No files matched your search
+163
-178
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+248
-243
@@ -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;
|
||||
+33
-44
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user