diff --git a/acs.coffee b/acs.coffee index 16ee813..d1216df 100644 --- a/acs.coffee +++ b/acs.coffee @@ -10,15 +10,14 @@ db = require './db' presets = require './presets' mongodb = require 'mongodb' fs = require 'fs' +apiFunctions = require './api-functions' applyConfigurations = (currentRequest, taskList) -> if taskList.length util.log("#{currentRequest.deviceId}: Presets discrepancy found") - db.tasksCollection.save(taskList, (err) -> - for task in taskList - util.log("#{currentRequest.deviceId}: Added #{task.name} task #{task._id}") - task = taskList[0] + apiFunctions.insertTasks(taskList, (err, tasks) -> + task = tasks[0] util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})") runTask(currentRequest, task, {}) ) @@ -99,9 +98,16 @@ updateDevice = (currentRequest, actions, callback) -> callback(err) if callback? return - task = {device : currentRequest.deviceId, name : 'init', timestamp : mongodb.Timestamp()} - db.tasksCollection.save(task, (err) -> - util.log("#{currentRequest.deviceId}: Added init task #{task._id}") + task = {device : currentRequest.deviceId, name : 'init', timestamp : config.INIT_TIMESTAMP} + apiFunctions.insertTasks(task, (err, t) -> + callback(err) if callback? + ) + ) + else if updates['_lastBootstrap']? + # reinitialize on bootstrap event (e.g. firmware upgrade) + db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$unset' : 'InternetGatewayDevice'}, {safe: true}, (err, count) -> + task = {device : currentRequest.deviceId, name : 'init', timestamp : config.INIT_TIMESTAMP} + apiFunctions.insertTasks(task, (err) -> callback(err) if callback? ) ) diff --git a/api-functions.coffee b/api-functions.coffee new file mode 100644 index 0000000..581655c --- /dev/null +++ b/api-functions.coffee @@ -0,0 +1,136 @@ +config = require './config' +db = require './db' +http = require 'http' +common = require './common' +util = require 'util' + + +connectionRequest = (deviceId, callback) -> + db.devicesCollection.findOne({_id : deviceId}, {'InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value' : 1}, (err, device)-> + if err + callback(err) + return + connectionRequestUrl = device.InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value + # for testing + #connectionRequestUrl = connectionRequestUrl.replace(/^(http:\/\/)([0-9\.]+)(\:[0-9]+\/[a-zA-Z0-9]+\/?$)/, '$110.1.1.254$3') + request = http.get(connectionRequestUrl, (res) -> + callback() + ) + + request.on('error', (err) -> + # error event when request is aborted + request.abort() + callback(err) + ) + + request.on('socket', (socket) -> + socket.setTimeout(2000) + socket.on('timeout', () -> + request.abort() + ) + ) + ) + + +watchTask = (taskId, timeout, callback) -> + setTimeout( () -> + db.tasksCollection.findOne({_id : taskId}, {'_id' : 1}, (err, task) -> + if task + timeout -= 500 + if timeout < 0 + callback('timeout') + else + watchTask(taskId, timeout, callback) + else + callback(err) + ) + , 500) + + +expandParam = (param) -> + params = [param] + for a,aa of config.ALIASES + if a == param or common.startsWith(a, "#{param}.") + for p in aa + params.push(p) if p[p.lastIndexOf('.') + 1] != '_' + + return params + + +sanitizeTask = (task, callback) -> + task.timestamp = new Date() + switch task.name + when 'getParameterValues' + projection = {} + for p in task.parameterNames + for pp in expandParam(p) + projection[pp] = 1 + db.devicesCollection.findOne({_id : task.device}, projection, (err, device) -> + parameterNames = [] + for k of projection + if common.getParamValueFromPath(device, k)? + parameterNames.push(k) + task.parameterNames = parameterNames + callback(task) + ) + when 'setParameterValues' + projection = {} + values = {} + for p in task.parameterValues + for pp in expandParam(p[0]) + projection[pp] = 1 + values[pp] = p[1] + db.devicesCollection.findOne({_id : task.device}, projection, (err, device) -> + parameterValues = [] + for k of projection + param = common.getParamValueFromPath(device, k) + if param? + parameterValues.push([k, values[k], param._type]) + task.parameterValues = parameterValues + callback(task) + ) + else + # TODO implement setParameterValues + callback(task) + + +addAliases = (device) -> + for k,v of config.ALIASES + for p in v + pp = p.split('.') + obj = device + for i in pp + if not obj[i]? + obj = null + break + obj = obj[i] + + device[k] = obj if obj? + + +insertTasks = (tasks, callback) -> + if common.typeOf(tasks) isnt common.ARRAY_TYPE + tasks = [tasks] + + counter = tasks.length + + for task in tasks + sanitizeTask(task, (t) -> + if t.uniqueKey? + db.tasksCollection.remove({device : t.device, uniqueKey : t.uniqueKey}, (err, removed) -> + ) + + --counter + if counter == 0 + db.tasksCollection.insert(tasks, (err, _tasks) -> + #util.log("#{_task.device}: Added task #{_task.name}(#{_task._id})") for _task in _tasks + callback(err, _tasks) + ) + ) + + +exports.addAliases = addAliases +exports.sanitizeTask = sanitizeTask +exports.connectionRequest = connectionRequest +exports.watchTask = watchTask +exports.insertTasks = insertTasks diff --git a/api.coffee b/api.coffee index cee87fb..bdd3185 100644 --- a/api.coffee +++ b/api.coffee @@ -7,6 +7,7 @@ db = require './db' mongodb = require 'mongodb' querystring = require 'querystring' query = require './query' +apiFunctions = require './api-functions' # regular expression objects DEVICE_TASKS_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/tasks\/?$/ @@ -17,108 +18,6 @@ FILES_REGEX = /^\/files\/([a-zA-Z0-9\-\_\%\ \.\/\(\)]+)\/?$/ PING_REGEX = /^\/ping\/([a-zA-Z0-9\-\_\.]+)\/?$/ QUERY_REGEX = /^\/([a-zA-Z0-9_]+s)\/?$/ -connectionRequest = (deviceId, callback) -> - db.devicesCollection.findOne({_id : deviceId}, {'InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value' : 1}, (err, device)-> - if err - callback(err) - return - connectionRequestUrl = device.InternetGatewayDevice.ManagementServer.ConnectionRequestURL._value - # for testing - #connectionRequestUrl = connectionRequestUrl.replace(/^(http:\/\/)([0-9\.]+)(\:[0-9]+\/[a-zA-Z0-9]+\/?$)/, '$1192.168.1.1$3') - request = http.get(connectionRequestUrl, (res) -> - callback() - ) - - request.on('error', (err) -> - # error event when request is aborted - request.abort() - callback(err) - ) - - request.on('socket', (socket) -> - socket.setTimeout(2000) - socket.on('timeout', () -> - request.abort() - ) - ) - ) - -watchTask = (taskId, timeout, callback) -> - setTimeout( () -> - db.tasksCollection.findOne({_id : taskId}, {'_id' : 1}, (err, task) -> - if task - timeout -= 500 - if timeout < 0 - callback('timeout') - else - watchTask(taskId, timeout, callback) - else - callback(err) - ) - , 500) - - -expandParam = (param) -> - params = [param] - for a,aa of config.ALIASES - if a == param or common.startsWith(a, "#{param}.") - for p in aa - params.push(p) if p[p.lastIndexOf('.') + 1] != '_' - - return params - - -sanitizeTask = (deviceId, task, callback) -> - task.device = deviceId - task.timestamp = new Date() - switch task.name - when 'getParameterValues' - projection = {} - for p in task.parameterNames - for pp in expandParam(p) - projection[pp] = 1 - db.devicesCollection.findOne({_id : deviceId}, projection, (err, device) -> - parameterNames = [] - for k of projection - if common.getParamValueFromPath(device, k)? - parameterNames.push(k) - task.parameterNames = parameterNames - callback(task) - ) - when 'setParameterValues' - projection = {} - values = {} - for p in task.parameterValues - for pp in expandParam(p[0]) - projection[pp] = 1 - values[pp] = p[1] - db.devicesCollection.findOne({_id : deviceId}, projection, (err, device) -> - parameterValues = [] - for k of projection - param = common.getParamValueFromPath(device, k) - if param? - parameterValues.push([k, values[k], param._type]) - task.parameterValues = parameterValues - callback(task) - ) - else - # TODO implement setParameterValues - callback(task) - - -addAliases = (device) -> - for k,v of config.ALIASES - for p in v - pp = p.split('.') - obj = device - for i in pp - if not obj[i]? - obj = null - break - obj = obj[i] - - device[k] = obj if obj? - cluster = require 'cluster' numCPUs = require('os').cpus().length @@ -220,40 +119,36 @@ else if request.method == 'POST' deviceId = querystring.unescape(DEVICE_TASKS_REGEX.exec(urlParts.pathname)[1]) if body - # queue given task - sanitizeTask(deviceId, JSON.parse(body), (task) -> - if task.uniqueKey? - db.tasksCollection.remove({device : deviceId, uniqueKey : task.uniqueKey}, (err, removed) -> - ) - db.tasksCollection.save(task, (err) -> - if err - response.writeHead(500) - response.end(err) - return + t = JSON.parse(body) + t.device = deviceId + apiFunctions.insertTasks(t, (err) -> + if err + response.writeHead(500) + response.end(err) + return - if urlParts.query.connection_request? - connectionRequest(deviceId, (err) -> - if err - response.writeHead(202) - response.end() - else - watchTask(task._id, config.DEVICE_ONLINE_THRESHOLD, (err) -> - if err - response.writeHead(202) - response.end() - return - response.writeHead(200) + 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() - ) - ) - else - 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 - connectionRequest(deviceId, (err) -> + apiFunctions.connectionRequest(deviceId, (err) -> if err response.writeHead 504 response.end() @@ -382,7 +277,7 @@ else response.end("\n]") else response.write(",\n") if i++ - addAliases(item) if collectionName is 'devices' + apiFunctions.addAliases(item) if collectionName is 'devices' response.write(JSON.stringify(item)) ) ) diff --git a/config.coffee b/config.coffee index 0e1c1a7..c7588e6 100644 --- a/config.coffee +++ b/config.coffee @@ -1,3 +1,4 @@ +exports.INIT_TIMESTAMP = new Date(1983, 9, 16) # arbitrary date exports.DATABASE_NAME = 'genie' exports.ACS_PORT = 7547 # CWMP port as assigned by IANA exports.ACS_HTTPS_PORT = 7548