Support for regular expressions for paths in parameter configuration.

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) ->
+        db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) ->
-          presets.getDevicePreset(currentRequest.deviceId, allPresets, allObjects, (devicePreset) ->
+          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()
                   )

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) ->
+addAliases = (device, aliases) ->
-  for k,v of config.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) ->
         )

api.coffee

@@ -164,33 +164,35 @@ else
           if body
             task = JSON.parse(body)
             task.device = deviceId
-            apiFunctions.insertTasks(task, (err) ->
+            db.getAliases((aliases) ->
-              db.redisClient.del("#{deviceId}_presets_hash", (err) ->
+              apiFunctions.insertTasks(task, aliases, (err) ->
-                throw err if 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()
-                    )
                 )
-              else
+                if err
-                response.writeHead(202)
+                  response.writeHead(500)
-                response.end()
+                  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) ->
+          db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) ->
-            presets.getDevicePreset(deviceId, allPresets, allObjects, (devicePreset) ->
+            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?
+        func = (aliases) ->
-          q = JSON.parse(querystring.unescape(urlParts.query.query))
+          if urlParts.query.query?
-        else
+            q = JSON.parse(querystring.unescape(urlParts.query.query))
-          q = {}
+          else
-        q = query.expand(q) if collectionName is 'devices'
+            q = {}
+          q = query.expand(q, aliases) if collectionName is 'devices'
 
-        if urlParts.query.projection?
+          if urlParts.query.projection?
-          projection = {}
+            projection = {}
-          for p in urlParts.query.projection.split(',')
+            for p in urlParts.query.projection.split(',')
-            p = p.trim()
+              p = p.trim()
-            projection[p] = 1
+              projection[p] = 1
-            if collectionName is 'devices'
+              if collectionName is 'devices'
-              for k,v of config.ALIASES
+                for k,v of aliases
-                if k == p or common.startsWith(k, "#{p}.")
+                  if k == p or common.startsWith(k, "#{p}.")
-                  projection[a] = 1 for a in v
+                    projection[a] = 1 for a in v
 
-        cur = collection.find(q, projection, {batchSize : 50})
+          cur = collection.find(q, projection, {batchSize : 50})
-        if urlParts.query.sort?
+          if urlParts.query.sort?
-          s = JSON.parse(querystring.unescape(urlParts.query.sort))
+            s = JSON.parse(querystring.unescape(urlParts.query.sort))
-          sort = {}
+            sort = {}
-          for k,v of s
+            for k,v of s
-            if config.ALIASES[k]?
+              if aliases[k]?
-              sort[a] = v for a in config.ALIASES[k]
+                sort[a] = v for a in aliases[k]
-            else
+              else
-              sort[k] = v
+                sort[k] = v
-          cur.sort(sort)
+            cur.sort(sort)
-        
+
-        cur.skip(parseInt(urlParts.query.skip)) if urlParts.query.skip?
+          cur.skip(parseInt(urlParts.query.skip)) if urlParts.query.skip?
-        cur.limit(parseInt(urlParts.query.limit)) if urlParts.query.limit?
+          cur.limit(parseInt(urlParts.query.limit)) if urlParts.query.limit?
-        cur.count((err, total) ->
+          cur.count((err, total) ->
-          response.writeHead(200, {'Content-Type' : 'application/json', 'total' : total})
+            response.writeHead(200, {'Content-Type' : 'application/json', 'total' : total})
-          if request.method is 'HEAD'
+            if request.method is 'HEAD'
-            response.end()
+              response.end()
-            return
+              return
-          response.write("[\n")
+            response.write("[\n")
-          i = 0
+            i = 0
-          cur.each((err, item) ->
+            cur.each((err, item) ->
-            if item is null
+              if item is null
-              response.end("\n]")
+                response.end("\n]")
-            else
+              else
-              response.write(",\n") if i++
+                response.write(",\n") if i++
-              apiFunctions.addAliases(item) if collectionName is 'devices'
+                apiFunctions.addAliases(item, aliases) if collectionName is 'devices'
-              response.write(JSON.stringify(item))
+                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')

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) ->
+getAliases = (callback) ->
-  redisClient.mget('presets', 'objects', (err, res) ->
+  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
+    aliases = JSON.parse(res[2])
-      callback(presets, objects)
+
-      return
+    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

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?
+  type = parameters.getType(path) ? 'default'
-    return normalizers[config.PARAMETERS[path].type](value, normType)
+  normalizers[type](value, normType)
-  else
-    return normalizers['default'](value, normType)

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

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

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]?
+  if aliases[param]?
-    for p in config.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})