diff --git a/acs.coffee b/acs.coffee index d62edef..dd4e2ca 100644 --- a/acs.coffee +++ b/acs.coffee @@ -20,20 +20,6 @@ INSTANCE_REGEX = /\.[\d]+\.$/ holdUntil = Date.now() -applyConfigurations = (currentRequest, taskList) -> - if taskList.length - util.log("#{currentRequest.deviceId}: Presets discrepancy found") - apiFunctions.insertTasks(taskList, (err, tasks) -> - throw new Error(err) if err? - task = tasks[0] - util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})") - runTask(currentRequest, task, {}) - ) - else - res = tr069.response(null, {}, {}) - writeResponse(currentRequest, res) - - writeResponse = (currentRequest, res) -> if res.headers['Content-Length'] == 0 # no more requests. terminate TCP connection @@ -122,14 +108,14 @@ updateDevice = (currentRequest, actions, callback) -> return task = {device : currentRequest.deviceId, name : 'init', timestamp : config.INIT_TIMESTAMP} - apiFunctions.insertTasks(task, (err, t) -> + apiFunctions.insertTasks(task, {}, (err, t) -> callback(err) if callback? ) ) else if updates['_lastBootstrap']? # reinitialize on bootstrap event (e.g. firmware upgrade) task = {device : currentRequest.deviceId, name : 'init', timestamp : config.INIT_TIMESTAMP} - apiFunctions.insertTasks(task, (err) -> + apiFunctions.insertTasks(task, {}, (err) -> callback(err) if callback? ) else @@ -209,14 +195,14 @@ assertPresets = (currentRequest) -> res = tr069.response(null, {}) writeResponse(currentRequest, res) else - db.getPresets((allPresets, allObjects) -> + db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) -> if not presetsHash? presetsHash = presets.calculatePresetsHash(allPresets, allObjects) db.redisClient.setex('presets_hash', config.PRESETS_CACHE_DURATION, presetsHash, (err, res) -> throw err if err ) - presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, (devicePreset) -> + presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, allAliases, (devicePreset) -> presets.processDevicePreset(currentRequest.deviceId, devicePreset, (taskList, addTags, deleteTags, expiry) -> db.redisClient.setex("#{currentRequest.deviceId}_presets_hash", Math.floor(Math.max(1, expiry - config.PRESETS_TIME_PADDING)), presetsHash, (err, res) -> throw err if err @@ -240,7 +226,7 @@ assertPresets = (currentRequest) -> if taskList.length t.expiry = expiry for t in taskList - apiFunctions.insertTasks(taskList, (err, taskList) -> + apiFunctions.insertTasks(taskList, allAliases, (err, taskList) -> throw err if err task = taskList[0] util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})") @@ -358,8 +344,8 @@ listener = (httpRequest, httpResponse) -> # Only supporting firmware upgrade for now return requestDownloadResponse() - db.getPresets((allPresets, allObjects) -> - presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, (devicePreset) -> + db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) -> + presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, allAliases, (devicePreset) -> presetSoftwareVersion = devicePreset.softwareVersion?.preset currentSoftwareVersion = devicePreset.softwareVersion?.current._value if presetSoftwareVersion? and presetSoftwareVersion != currentSoftwareVersion @@ -381,7 +367,7 @@ listener = (httpRequest, httpResponse) -> name : 'download', file : file['_id'] } - apiFunctions.insertTasks(task, (err, tasks) -> + apiFunctions.insertTasks(task, allAliases, (err, tasks) -> throw err if err return requestDownloadResponse() ) diff --git a/api-functions.coffee b/api-functions.coffee index 6fc41bd..1fe2a47 100644 --- a/api-functions.coffee +++ b/api-functions.coffee @@ -73,9 +73,9 @@ watchTask = (taskId, timeout, callback) -> , 500) -expandParam = (param) -> +expandParam = (param, aliases) -> params = [param] - for a,aa of config.ALIASES + for a,aa of aliases if a == param or common.startsWith(a, "#{param}.") for p in aa params.push(p) if p[p.lastIndexOf('.') + 1] != '_' @@ -83,7 +83,7 @@ expandParam = (param) -> return params -sanitizeTask = (task, callback) -> +sanitizeTask = (task, aliases, callback) -> task.timestamp = new Date(task.timestamp ? Date.now()) if task.expiry? if common.typeOf(task.expiry) is common.DATE_TYPE or isNaN(task.expiry) @@ -95,7 +95,7 @@ sanitizeTask = (task, callback) -> when 'getParameterValues' projection = {} for p in task.parameterNames - for pp in expandParam(p) + for pp in expandParam(p, aliases) projection[pp] = 1 db.devicesCollection.findOne({_id : task.device}, projection, (err, device) -> parameterNames = [] @@ -109,7 +109,7 @@ sanitizeTask = (task, callback) -> projection = {} values = {} for p in task.parameterValues - for pp in expandParam(p[0]) + for pp in expandParam(p[0], aliases) projection[pp] = 1 values[pp] = p[1] db.devicesCollection.findOne({_id : task.device}, projection, (err, device) -> @@ -126,8 +126,8 @@ sanitizeTask = (task, callback) -> callback(task) -addAliases = (device) -> - for k,v of config.ALIASES +addAliases = (device, aliases) -> + for k,v of aliases for p in v pp = p.split('.') obj = device @@ -140,14 +140,14 @@ addAliases = (device) -> device[k] = obj if obj? -insertTasks = (tasks, callback) -> +insertTasks = (tasks, aliases, callback) -> if common.typeOf(tasks) isnt common.ARRAY_TYPE tasks = [tasks] counter = tasks.length for task in tasks - sanitizeTask(task, (t) -> + sanitizeTask(task, aliases, (t) -> if t.uniqueKey? db.tasksCollection.remove({device : t.device, uniqueKey : t.uniqueKey}, (err, removed) -> ) diff --git a/api.coffee b/api.coffee index f22c080..1ddb5a5 100644 --- a/api.coffee +++ b/api.coffee @@ -164,33 +164,35 @@ else if body task = JSON.parse(body) task.device = deviceId - apiFunctions.insertTasks(task, (err) -> - db.redisClient.del("#{deviceId}_presets_hash", (err) -> - throw err if err - ) - if err - response.writeHead(500) - response.end(err) - return - - if urlParts.query.connection_request? - apiFunctions.connectionRequest(deviceId, (err) -> - if err - response.writeHead(202) - response.end() - else - apiFunctions.watchTask(task._id, config.DEVICE_ONLINE_THRESHOLD, (err) -> - if err - response.writeHead(202) - response.end() - return - response.writeHead(200) - response.end() - ) + db.getAliases((aliases) -> + apiFunctions.insertTasks(task, aliases, (err) -> + db.redisClient.del("#{deviceId}_presets_hash", (err) -> + throw err if err ) - else - response.writeHead(202) - response.end() + if err + response.writeHead(500) + response.end(err) + return + + if urlParts.query.connection_request? + apiFunctions.connectionRequest(deviceId, (err) -> + if err + response.writeHead(202) + response.end() + else + apiFunctions.watchTask(task._id, config.DEVICE_ONLINE_THRESHOLD, (err) -> + if err + response.writeHead(202) + response.end() + return + response.writeHead(200) + response.end() + ) + ) + else + response.writeHead(202) + response.end() + ) ) else if urlParts.query.connection_request? # no task, send connection request only @@ -211,8 +213,8 @@ else else if DEVICE_PRESET_REGEX.test(urlParts.pathname) deviceId = querystring.unescape(DEVICE_PRESET_REGEX.exec(urlParts.pathname)[1]) if request.method is 'GET' - db.getPresets((allPresets, allObjects) -> - presets.getDevicePreset(deviceId, allPresets, allObjects, (devicePreset) -> + db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) -> + presets.getDevicePreset(deviceId, allPresets, allObjects, allAliases, (devicePreset) -> response.writeHead(200, {'Content-Type' : 'application/json'}) response.end(JSON.stringify(devicePreset)) ) @@ -301,51 +303,59 @@ else response.end('404 Not Found') return - if urlParts.query.query? - q = JSON.parse(querystring.unescape(urlParts.query.query)) - else - q = {} - q = query.expand(q) if collectionName is 'devices' + func = (aliases) -> + if urlParts.query.query? + q = JSON.parse(querystring.unescape(urlParts.query.query)) + else + q = {} + q = query.expand(q, aliases) if collectionName is 'devices' - if urlParts.query.projection? - projection = {} - for p in urlParts.query.projection.split(',') - p = p.trim() - projection[p] = 1 - if collectionName is 'devices' - for k,v of config.ALIASES - if k == p or common.startsWith(k, "#{p}.") - projection[a] = 1 for a in v + if urlParts.query.projection? + projection = {} + for p in urlParts.query.projection.split(',') + p = p.trim() + projection[p] = 1 + if collectionName is 'devices' + for k,v of aliases + if k == p or common.startsWith(k, "#{p}.") + projection[a] = 1 for a in v - cur = collection.find(q, projection, {batchSize : 50}) - if urlParts.query.sort? - s = JSON.parse(querystring.unescape(urlParts.query.sort)) - sort = {} - for k,v of s - if config.ALIASES[k]? - sort[a] = v for a in config.ALIASES[k] - else - sort[k] = v - cur.sort(sort) - - cur.skip(parseInt(urlParts.query.skip)) if urlParts.query.skip? - cur.limit(parseInt(urlParts.query.limit)) if urlParts.query.limit? - cur.count((err, total) -> - response.writeHead(200, {'Content-Type' : 'application/json', 'total' : total}) - if request.method is 'HEAD' - response.end() - return - response.write("[\n") - i = 0 - cur.each((err, item) -> - if item is null - response.end("\n]") - else - response.write(",\n") if i++ - apiFunctions.addAliases(item) if collectionName is 'devices' - response.write(JSON.stringify(item)) + cur = collection.find(q, projection, {batchSize : 50}) + if urlParts.query.sort? + s = JSON.parse(querystring.unescape(urlParts.query.sort)) + sort = {} + for k,v of s + if aliases[k]? + sort[a] = v for a in aliases[k] + else + sort[k] = v + cur.sort(sort) + + cur.skip(parseInt(urlParts.query.skip)) if urlParts.query.skip? + cur.limit(parseInt(urlParts.query.limit)) if urlParts.query.limit? + cur.count((err, total) -> + response.writeHead(200, {'Content-Type' : 'application/json', 'total' : total}) + if request.method is 'HEAD' + response.end() + return + response.write("[\n") + i = 0 + cur.each((err, item) -> + if item is null + response.end("\n]") + else + response.write(",\n") if i++ + apiFunctions.addAliases(item, aliases) if collectionName is 'devices' + response.write(JSON.stringify(item)) + ) ) - ) + + if collectionName is 'devices' + db.getAliases((aliases) -> + func(aliases) + ) + else + func({}) else response.writeHead 404 response.end('404 Not Found') diff --git a/db.coffee b/db.coffee index ee7b0ad..7a4e621 100644 --- a/db.coffee +++ b/db.coffee @@ -1,6 +1,7 @@ config = require './config' mongodb = require 'mongodb' redisClient = require('redis').createClient(config.REDIS_SOCKET) +parameters = require './parameters' tasksCollection = null devicesCollection = null @@ -98,13 +99,21 @@ getCached = (name, valueCallback, valueExpiry, callback) -> ) -getPresets = (callback) -> - redisClient.mget('presets', 'objects', (err, res) -> +getAliases = (callback) -> + getCached('aliases', parameters.compileAliases, config.PRESETS_CACHE_DURATION, (err, res) -> + throw err if err + callback(res) + ) + + +getPresetsObjectsAliases = (callback) -> + redisClient.mget('presets', 'objects', 'aliases', (err, res) -> presets = JSON.parse(res[0]) objects = JSON.parse(res[1]) - if presets and objects - callback(presets, objects) - return + aliases = JSON.parse(res[2]) + + if presets and objects and aliases + return callback(presets, objects, aliases) if not presets getCached('presets', (callback) -> @@ -114,7 +123,7 @@ getPresets = (callback) -> , config.PRESETS_CACHE_DURATION, (err, res) -> throw err if err presets = res - callback(presets, objects) if objects + callback(presets, objects, aliases) if objects and aliases ) if not objects @@ -128,11 +137,19 @@ getPresets = (callback) -> , config.PRESETS_CACHE_DURATION, (err, res) -> throw err if err objects = res - callback(presets, objects) if presets + callback(presets, objects, aliases) if presets and aliases + ) + + if not aliases + getCached('aliases', parameters.compileAliases, config.PRESETS_CACHE_DURATION, (err, res) -> + throw err if err + aliases = res + callback(presets, objects, aliases) if presets and objects ) ) exports.redisClient = redisClient exports.getTask = getTask -exports.getPresets = getPresets +exports.getPresetsObjectsAliases = getPresetsObjectsAliases +exports.getAliases = getAliases diff --git a/normalize.coffee b/normalize.coffee index 729cb90..71a2de3 100644 --- a/normalize.coffee +++ b/normalize.coffee @@ -1,5 +1,5 @@ -config = require './config' common = require './common' +parameters = require './parameters' stringToRegexp = (input) -> @@ -105,7 +105,5 @@ normalizers.mac = (input, normType) -> exports.normalize = (path, value, normType) -> - if path of config.PARAMETERS and config.PARAMETERS[path].type? - return normalizers[config.PARAMETERS[path].type](value, normType) - else - return normalizers['default'](value, normType) + type = parameters.getType(path) ? 'default' + normalizers[type](value, normType) diff --git a/parameters.coffee b/parameters.coffee new file mode 100644 index 0000000..011ba2d --- /dev/null +++ b/parameters.coffee @@ -0,0 +1,179 @@ +config = require './config' +db = require './db' + + +PATH_REGEX = /\w+|(\/.+?\/\w*)/g + +# Javascript represents numbers as 64-bit floating point numbers +# Integers up to 2^53 can be encoded precisely +MAX_PARAMETERS = 52 + + +# get the index of the least significant bit +getLsb = (num) -> + i = -1 + while num > 0 + break if num & 1 + num >>= 1 + ++ i + return i + + +# compile types into data types optimized for quick lookup +types = [] +stringSegments = {} +regexSegments = {} +paramRegex = {} + +for k, v of config.PARAMETERS + continue if not v.type? + typeIndex = types.push(v.type) - 1 + throw new Error("There cannot be more than #{MAX_PARAMETERS} configured parameters") if typeIndex >= MAX_PARAMETERS + segmentIndex = 0 + parts = k.match(PATH_REGEX) + for i in [0 ... parts.length] + part = parts[i] + positionHash = i + parts.length * MAX_PARAMETERS + + if part[0] == '/' # this is a regex part + j = part.lastIndexOf('/') + regExp = new RegExp(part[1...j], part[j+1..]) + regexSegments[positionHash] ?= 0 + regexSegments[positionHash] |= Math.pow(2, typeIndex) + paramRegex[typeIndex] ?= {} + paramRegex[typeIndex][i] = regExp + else + stringSegments[positionHash] ?= {} + stringSegments[positionHash][part] ?= 0 + stringSegments[positionHash][part] |= Math.pow(2, typeIndex) + + +getType = (param) -> + parts = param.split('.') + positionHash = parts.length * MAX_PARAMETERS + + stringIndices = 0 | stringSegments[positionHash]?[parts[0]] + allIndices = stringIndices | regexSegments[positionHash] + + for i in [1...parts.length] + positionHash = i + parts.length * MAX_PARAMETERS + strIdx = stringSegments[positionHash]?[parts[i]] + rgxIdx = regexSegments[positionHash] + stringIndices &= strIdx + allIndices &= (strIdx | rgxIdx) + + if stringIndices > 0 + return types[getLsb(stringIndices)] + + typeIndex = 0 + while allIndices > 0 + if allIndices & 1 + match = true + for i, r of paramRegex[typeIndex] + if not r.test(parts[i]) + match = false + break + + if match + return types[typeIndex] + + ++ typeIndex + allIndices >>= 1 + + return null + + +splitIds = (batches, callback) -> + ids = [{}] + return callback(ids) if batches == 1 + db.devicesCollection.count((err, count) -> + return callback(ids) if count <= 1000 + batchSize = Math.floor(count / batches) + for i in [1 ... batches] + ids[i] = {} + do (i) -> + cursor = db.devicesCollection.find({}, {_id:1}).sort({'_id':1}).skip(i * batchSize).limit(1) + cursor.nextObject((err, obj) -> + ids[i]['$gte'] = obj._id + ids[i-1]['$lt'] = obj._id + if i == batches - 1 + callback(ids) + ) + ) + + +compileAliases = (callback) -> + map = () -> + for key, parts of keys + ref = this + path = '' + for part in parts + if part.test? + r = ref + ref = null + for p of r + if part.test(p) + path += ".#{p}" + ref = r[p] + break + else + path += ".#{part}" + ref = ref[part] + break if not ref + emit(path[1..], key) if ref + return + + reduce = (k, v) -> + return v[0] + + aliases = {} + keys = {} + for k, v of config.PARAMETERS + alias = v.alias + continue if not alias? + isStatic = true + ar = [] + for p in k.match(/\w+|(\/.+?\/\w*)/g) + if p[0] == '/' # this is a regex + isStatic = false + i = p.lastIndexOf('/') + r = new RegExp(p[1...i], p[i+1..]) + ar.push(r) + else + ar.push(p) + + if isStatic + if not aliases[alias]? + aliases[alias] = [] + aliases[alias].push(k) + else + keys[k] = ar + + counter = 0 + splitIds(4, (batches) -> + for batch in batches + options = { + out : {inline : 1} + jsMode : true + sort : {$natural : 1} + scope : {keys : keys, flags : {}} + query : {_id : batch} if batches.length > 1 + } + + db.devicesCollection.mapReduce(map, reduce, options, (err, out) -> + for o in out + alias = config.PARAMETERS[o.value].alias + if not aliases[alias]? + aliases[alias] = [] + id = String(o._id) + if id not in aliases[alias] + aliases[alias].push(id) + + if ++counter == batches.length + callback(err, aliases, config.ALIASES_CACHE) + ) + ) + + +exports.getType = getType +exports.compileAliases = compileAliases diff --git a/presets.coffee b/presets.coffee index 747d17d..78d9ec1 100644 --- a/presets.coffee +++ b/presets.coffee @@ -22,12 +22,12 @@ matchObject = (object, param) -> return true -exports.getDevicePreset = (deviceId, presets, objects, callback) -> +exports.getDevicePreset = (deviceId, presets, objects, aliases, callback) -> deviceCustomCommands = customCommands.getDeviceCustomCommands(deviceId) # only fetch relevant params projection = {_id : 1} for p in presets - p.precondition = query.expand(p.precondition) + p.precondition = query.expand(p.precondition, aliases) mongoQuery.projection(p.precondition, projection) for c in p.configurations diff --git a/query.coffee b/query.coffee index aa34d5a..5d766d8 100644 --- a/query.coffee +++ b/query.coffee @@ -1,4 +1,3 @@ -config = require './config' common = require './common' normalize = require './normalize' @@ -40,10 +39,10 @@ expandValue = (param, value) -> return objs -permute = (param, val) -> +permute = (param, val, aliases) -> keys = [] - if config.ALIASES[param]? - for p in config.ALIASES[param] + if aliases[param]? + for p in aliases[param] keys.push(p) else keys.push(param) @@ -62,16 +61,16 @@ permute = (param, val) -> return conditions -expand = (query) -> +expand = (query, aliases) -> new_query = {} for k,v of query if k[0] == '$' # operator expressions = [] for e in v - expressions.push(expand(e)) + expressions.push(expand(e, aliases)) new_query[k] = expressions else - conditions = permute(k, v) + conditions = permute(k, v, aliases) if conditions.length > 1 if new_query['$and']? new_query['$and'].push({'$or' : conditions})