diff --git a/acs.coffee b/acs.coffee index 37a06b1..346c5ba 100644 --- a/acs.coffee +++ b/acs.coffee @@ -6,18 +6,41 @@ tr069 = require './tr-069' tasks = require './tasks' sanitize = require('./sanitize').sanitize db = require './db' +profiles = require './profiles' -currentClientIP = null +currentRequest = null -writeResponse = (serverResponse, res) -> + +applyConfigurations = (configurations) -> + taskList = [] + for c of configurations + task = {device : currentRequest.deviceId, name : 'setParameterValues', parameterValues: [], timestamp : db.mongo.Timestamp()} + task.parameterValues.push([c, configurations[c]]) + taskList.push task + + if taskList.length + util.log("#{currentRequest.deviceId}: Profiles discrepancy found") + db.tasksCollection.save(taskList, (err) -> + for task in taskList + util.log("#{currentRequest.deviceId}: Added #{task.name} task #{task._id}") + task = taskList[0] + util.log("#{currentRequest.deviceId}: started task #{task.name}(#{task._id})") + runTask(task, {}) + ) + else + res = tr069.response(currentRequest.sessionId, {}, {}) + writeResponse res + + +writeResponse = (res) -> if config.DEBUG s = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n" - db.memcached.append("debug-#{currentClientIP}", s, (err, result) -> + db.memcached.append("debug-#{currentRequest.clientIp}", s, (err, result) -> ) - serverResponse.writeHead(res.code, res.headers) - serverResponse.end(res.data) + currentRequest.response.writeHead(res.code, res.headers) + currentRequest.response.end(res.data) updateDevice = (deviceId, actions, callback) -> @@ -50,21 +73,21 @@ updateDevice = (deviceId, actions, callback) -> updates["#{path}_timestamp"] = now if Object.keys(updates).length > 0 - db.devicesCollection.update({'_id' : deviceId}, {'$set' : updates}, {safe: true}, (err, count) -> + db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {safe: true}, (err, count) -> if (err) callback(err) if callback? return if count == 0 - util.log("#{deviceId}: new device detected") - db.devicesCollection.update({'_id' : deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) -> + util.log("#{currentRequest.deviceId}: new device detected") + db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true, safe: true}, (err) -> if err? callback(err) if callback? return - task = {device : deviceId, name : 'init', timestamp : db.mongo.Timestamp()} + task = {device : currentRequest.deviceId, name : 'init', timestamp : db.mongo.Timestamp()} db.tasksCollection.save(task, (err) -> - util.log("#{deviceId}: Added init task #{task._id}") + util.log("#{currentRequest.deviceId}: Added init task #{task._id}") callback(err) if callback? ) ) @@ -75,39 +98,63 @@ updateDevice = (deviceId, actions, callback) -> callback() if callback? -runTask = (sessionId, deviceId, task, reqParams, response) -> +runTask = (task, reqParams) -> resParams = {} actions = tasks.task(task, reqParams, resParams) - updateDevice(deviceId, actions) + updateDevice(currentRequest.deviceId, actions) if Object.keys(resParams).length > 0 db.memcached.set(String(task._id), task, config.CACHE_DURATION, (err, result) -> - res = tr069.response(sessionId, resParams, {task : String(task._id)}) - writeResponse response, res + res = tr069.response(currentRequest.sessionId, resParams, {task : String(task._id)}) + writeResponse res ) return # Task finished db.memcached.del(String(task._id)) db.tasksCollection.remove({'_id' : db.mongo.ObjectID(task._id)}, {safe: true}, (err, removed) -> - util.log("#{deviceId}: completed task #{task.name}(#{task._id})") + util.log("#{currentRequest.deviceId}: completed task #{task.name}(#{task._id})") cookies = {task : null} - nextTask(sessionId, deviceId, response, cookies) + nextTask(cookies) ) -nextTask = (sessionId, deviceId, response, cookies) -> - cur = db.tasksCollection.find({'device' : deviceId}).sort(['timestamp']).limit(1) +nextTask = (cookies) -> + cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1) cur.nextObject( (err, task) -> reqParams = {} resParams = {} + if not task - res = tr069.response(sessionId, resParams, cookies) - writeResponse response, res + # no more taks, check profile discrepancy + db.memcached.get(["#{currentRequest.deviceId}_profiles_hash", 'profiles_hash'], (err, results) -> + profilesHash = results['profiles_hash'] + deviceProfilesHash = results["#{currentRequest.deviceId}_profiles_hash"] + + if not deviceProfilesHash or profilesHash != deviceProfilesHash + profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (configurations) -> + applyConfigurations(configurations) + ) + else if not profilesHash + profiles.getProfilesHash((hash) -> + if hash != deviceProfilesHash + profiles.findDeviceConfigurationDiscrepancy(currentRequest.deviceId, profilesHash, (configurations) -> + applyConfigurations(configurations) + ) + else + # no discrepancy, return empty response + res = tr069.response(currentRequest.sessionId, resParams, cookies) + writeResponse res + ) + else + # no discrepancy, return empty response + res = tr069.response(currentRequest.sessionId, resParams, cookies) + writeResponse res + ) return - util.log("#{deviceId}: started task #{task.name}(#{task._id})") - runTask(sessionId, deviceId, task, reqParams, response) + util.log("#{currentRequest.deviceId}: started task #{task.name}(#{task._id})") + runTask(task, reqParams) ) @@ -131,12 +178,13 @@ if cluster.isMaster else if config.DEBUG process.on('uncaughtException', (error) -> + util.error(error.stack) # dump request/response logs and stack trace - util.log("Unexpected error occured. Writing log to debug/#{currentClientIP}.log.") - db.memcached.get("debug-#{currentClientIP}", (err, l) -> + db.memcached.get("debug-#{currentRequest.clientIp}", (err, l) -> + util.log("Unexpected error occured. Writing log to debug/#{currentRequest.clientIp}.log.") util.error(err) if err fs = require 'fs' - fs.writeFileSync("debug/#{currentClientIP}.log", l + "\n\n" + error.stack) + fs.writeFileSync("debug/#{currentRequest.clientIp}.log", l + "\n\n" + error.stack) process.exit(1) ) ) @@ -169,12 +217,15 @@ else return body.toString(encoding || 'utf8', 0, body.byteLength) request.addListener 'end', () -> - currentClientIP = request.connection.remoteAddress + currentRequest = {} + currentRequest.request = request + currentRequest.response = response + currentRequest.clientIp = request.connection.remoteAddress if config.DEBUG s = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(request.headers) + "\n#{request.getBody()}\n\n" - db.memcached.append("debug-#{currentClientIP}", s, (err, result) -> + db.memcached.append("debug-#{currentRequest.clientIp}", s, (err, result) -> if err == 'Item is not stored' - db.memcached.set("debug-#{currentClientIP}", s, config.CACHE_DURATION, (err, result) -> + db.memcached.set("debug-#{currentRequest.clientIp}", s, config.CACHE_DURATION, (err, result) -> ) ) @@ -183,26 +234,27 @@ else if reqParams.inform resParams.inform = true - cookies.ID = sessionId = reqParams.sessionId - cookies.DeviceId = deviceId = common.getDeviceId(reqParams.deviceId) + cookies.ID = currentRequest.sessionId = reqParams.sessionId + cookies.DeviceId = currentRequest.deviceId = common.getDeviceId(reqParams.deviceId) if config.DEBUG - util.log("#{deviceId}: inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}") + util.log("#{currentRequest.deviceId}: inform (#{reqParams.eventCodes}); retry count #{reqParams.retryCount}") - updateDevice(deviceId, {'inform' : true, 'parameterValues' : reqParams.informParameterValues}, (err) -> - res = tr069.response(sessionId, resParams, cookies) - writeResponse response, res + updateDevice(currentRequest.deviceId, {'inform' : true, 'parameterValues' : reqParams.informParameterValues}, (err) -> + res = tr069.response(currentRequest.sessionId, resParams, cookies) + writeResponse res ) return - sessionId = reqParams.cookies.ID - deviceId = reqParams.cookies.DeviceId + currentRequest.sessionId = reqParams.cookies.ID + currentRequest.deviceId = reqParams.cookies.DeviceId taskId = reqParams.cookies.task if not taskId - nextTask(sessionId, deviceId, response, cookies) + nextTask(cookies) else db.memcached.get(taskId, (err, task) -> - runTask(sessionId, deviceId, task, reqParams, response) + # TODO if not found in memcached, fetch from mongo + runTask(task, reqParams) ) server.listen config.ACS_PORT, config.ACS_INTERFACE diff --git a/api.coffee b/api.coffee index 38f1e6e..496a48e 100644 --- a/api.coffee +++ b/api.coffee @@ -107,10 +107,28 @@ else body = request.getBody() urlParts = url.parse(request.url, true) if PROFILES_REGEX.test(urlParts.pathname) + profileName = PROFILES_REGEX.exec(urlParts.pathname)[1] if request.method == 'PUT' - # TODO save profile + profile = JSON.parse(body) + profile._id = profileName + + profilesCollection.save(profile, (err) -> + if err + response.writeHead(500) + response.end(err) + return + response.writeHead(200) + response.end() + ) else if request.method == 'DELETE' - # TODO delete profile + profilesCollection.remove({'_id' : profileName}, (err, removedCount) -> + if err + response.writeHead(500) + response.end(err) + return + response.writeHead(200) + response.end() + ) else response.writeHead 405, {'Allow': 'PUT, DELETE'} response.end('405 Method Not Allowed') diff --git a/config.coffee b/config.coffee index 8aa86ea..391608e 100644 --- a/config.coffee +++ b/config.coffee @@ -3,7 +3,8 @@ exports.ACS_PORT = 7547 # same as AP Manager's port exports.ACS_INTERFACE = '0.0.0.0' exports.API_PORT = 7557 exports.API_INTERFACE = '0.0.0.0' -exports.CACHE_DURATION = 60000 # in milliseconds +exports.CACHE_DURATION = 60 # in seconds +exports.PROFILES_CACHE_DURATION = 86400 exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock' exports.MEMCACHED_SOCKET = '/tmp/memcached.sock' exports.WORKER_RESPAWN_TIME = 3000 diff --git a/db.coffee b/db.coffee index 535e6a2..89bc1e6 100644 --- a/db.coffee +++ b/db.coffee @@ -1,13 +1,15 @@ config = require './config' mongo = require 'mongodb' Memcached = require 'memcached' - -exports.memcached = new Memcached(config.MEMCACHED_SOCKET) +Memcached.config.maxExpiration = 86400 +memcached = new Memcached(config.MEMCACHED_SOCKET) # Create MongoDB connections -dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true}) tasksCollection = null devicesCollection = null +profilesCollection = null + +dbserver = new mongo.Server(config.MONGODB_SOCKET, 0, {auto_reconnect: true}) db = new mongo.Db(config.DATABASE_NAME, dbserver, {native_parser:true, safe:true}) db.open( (err, db) -> @@ -20,6 +22,11 @@ db.open( (err, db) -> db.collection('devices', (err, collection) -> exports.devicesCollection = collection ) + + db.collection('profiles', (err, collection) -> + exports.profilesCollection = collection + ) ) -exports.mongo = mongo \ No newline at end of file +exports.mongo = mongo +exports.memcached = memcached diff --git a/profiles.coffee b/profiles.coffee new file mode 100644 index 0000000..896fc8a --- /dev/null +++ b/profiles.coffee @@ -0,0 +1,123 @@ +config = require './config' +db = require './db' + + +getProfiles = (callback) -> + db.memcached.get('profiles', (err, res) -> + throw err if err + + if res + callback(res) + return + + db.profilesCollection.find().toArray((err, res) -> + if err + callback(err, res) + return + + profiles = {} + for p in res + profiles[p['_id']] = p + + db.memcached.set('profiles', profiles, config.PROFILES_CACHE_DURATION) + callback(profiles) + ) + ) + + +getProfilesHash = (callback) -> + db.memcached('profiles_hash', (err, res) -> + if res + callback(res) + else + getProfiles((profiles) -> + hash = callback(calculateProfilesHash(profiles)) + db.memcached.set('profiles_hash', hash, config.PROFILES_CACHE_DURATION, (err, res) -> + callback(hash) + ) + ) + ) + + +calculateProfilesHash = (profiles) -> + crypto = require('crypto') + hash = crypto.createHash('md5').update(JSON.stringify(profiles)).digest('hex') + return hash + + +getProfilesVaraibles = (profiles) -> + vars = [] + for p in profiles + for c in p.configurations + vars.push(c[0]) + return vars + + +exports.findDeviceConfigurationDiscrepancy = (deviceId, profilesHash, callback) -> + getProfiles((profiles) -> + params = {'_tags' : 1} + for p of profiles + profile = profiles[p] + for c in profile.configurations + switch c.type + when 'value' + params[c.param] = 1 + else + throw 'Unknown configuration type' + + db.devicesCollection.findOne({'_id' : deviceId}, params, (err, device) -> + tags = device['_tags'] or [] + tags.push('default') + + deviceProfiles = [] + for t in tags + if profiles[t]? + deviceProfiles.push(profiles[t]) + + combinedConfigurations = combineProfileConfigurations(deviceProfiles) + + discrepency = {} + for c in combinedConfigurations + switch c.type + when 'value' + if c.value != getDeviceParamFromPath(device, "#{c.param}._value") + discrepency[c.param] = c.value + else + throw 'Unknown configuration type' + if not profilesHash + profilesHash = calculateProfilesHash(profiles) + db.memcached.set('profiles_hash', profilesHash, config.PROFILES_CACHE_DURATION, (err, res) -> + ) + + db.memcached.set("#{deviceId}_profiles_hash", profilesHash, config.PROFILES_CACHE_DURATION, (err, res) -> + callback(discrepency) + ) + ) + ) + + +combineProfileConfigurations = (profiles) -> + maxWeights = {} + configurations = {} + for p in profiles + for c in p.configurations + type = c.type + switch type + when 'value' + paramName = c.param + if not maxWeights["#{type}_#{paramName}"]? or p.weight > maxWeights["#{type}_#{paramName}"] + configurations["#{type}_#{paramName}"] = c + maxWeights["#{type}_#{paramName}"] = p.weight + else + throw 'Unknown configuration type' + + configurationsList = (configurations[c] for c of configurations) + return configurationsList + + +getDeviceParamFromPath = (device, paramPath) -> + path = paramPath.split('.') + ref = device + for p in path + ref = ref[p] + return ref \ No newline at end of file