rodrigo/genieacs
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Accept configuration from environment variables or command line arguments. Unified cluster and HTTP server code for CWMP, NBI, and FS.

Browse Source
This commit is contained in:
Zaid Abdulla committed 2014-07-31 01:44:53 +03:00
1 parent 5d0f398bf7
commit 32c56f8ff1
14 files changed
+511 -485
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

bin/genieacs-cwmp
+3 -4
View File
@@ -2,9 +2,8 @@
var path = require('path'); var path = require('path');
var fs = require('fs'); var fs = require('fs');
var dir = path.join(path.dirname(fs.realpathSync(__filename)), '../lib/'); var dir = path.resolve(path.dirname(fs.realpathSync(__filename)), '..');
process.title = 'genieacs-cwmp';
process.chdir(dir); process.chdir(dir);
require(path.join(dir, './cwmp')); cluster = require(path.resolve(dir, 'lib/cluster'));
cluster.start('cwmp');
bin/genieacs-fs
+3 -4
View File
@@ -2,9 +2,8 @@
var path = require('path'); var path = require('path');
var fs = require('fs'); var fs = require('fs');
var dir = path.join(path.dirname(fs.realpathSync(__filename)), '../lib/'); var dir = path.resolve(path.dirname(fs.realpathSync(__filename)), '..');
process.title = 'genieacs-fs';
process.chdir(dir); process.chdir(dir);
require(path.join(dir, './fs')); cluster = require(path.resolve(dir, 'lib/cluster'));
cluster.start('fs');
bin/genieacs-nbi
+3 -4
View File
@@ -2,9 +2,8 @@
var path = require('path'); var path = require('path');
var fs = require('fs'); var fs = require('fs');
var dir = path.join(path.dirname(fs.realpathSync(__filename)), '../lib/'); var dir = path.resolve(path.dirname(fs.realpathSync(__filename)), '..');
process.title = 'genieacs-nbi';
process.chdir(dir); process.chdir(dir);
require(path.join(dir, './nbi')); cluster = require(path.resolve(dir, 'lib/cluster'));
cluster.start('nbi');
config/config-sample.json
+11 -11
View File
@@ -1,15 +1,15 @@
{ {
"MONGODB_CONNECTION_URL" : "mongodb://localhost/genieacs", "MONGODB_CONNECTION_URL" : "mongodb://127.0.0.1/genieacs",
"REDIS_SOCKET" : "6379", "REDIS_PORT" : "6379",
"ACS_INTERFACE" : "0.0.0.0", "REDIS_HOST" : "127.0.0.1",
"ACS_PORT" : 7547, "CWMP_INTERFACE" : "0.0.0.0",
"ACS_HTTPS_INTERFACE" : "0.0.0.0", "CWMP_PORT" : 7547,
"ACS_HTTPS_PORT" : 7548, "CWMP_SSL" : false,
"API_INTERFACE" : "0.0.0.0", "NBI_INTERFACE" : "0.0.0.0",
"API_PORT" : 7557, "NBI_PORT" : 7557,
"FILES_INTERFACE" : "0.0.0.0", "FS_INTERFACE" : "0.0.0.0",
"FILES_PORT" : 7567, "FS_PORT" : 7567,
"FILES_IP" : "192.168.0.1", "FILES_IP" : "192.168.0.1",
"LOG_INFORMS" : true, "LOG_INFORMS" : true,
"DEBUG_DEVICES" : {} "DEBUG" : false
} }
config/httpscert-sample.crt
-17
View File
@@ -1,17 +0,0 @@
-----BEGIN CERTIFICATE-----
MIICqTCCAhICCQDSrolRw1qPBzANBgkqhkiG9w0BAQUFADCBmDELMAkGA1UEBhMC
SVExFTATBgNVBAgMDFN1bGF5bWFuaXlhaDEVMBMGA1UEBwwMU3VsYXltYW5peWFo
MQ8wDQYDVQQKDAZGYW5vb3MxEzARBgNVBAsMClRlY2hub2xvZ3kxDjAMBgNVBAMM
BWdlbmllMSUwIwYJKoZIhvcNAQkBFhZ6YWlkLmFiZHVsbGFAZmFub29zLmlxMB4X
DTEzMDIxNzEyNTYzN1oXDTE0MDIxNzEyNTYzN1owgZgxCzAJBgNVBAYTAklRMRUw
EwYDVQQIDAxTdWxheW1hbml5YWgxFTATBgNVBAcMDFN1bGF5bWFuaXlhaDEPMA0G
A1UECgwGRmFub29zMRMwEQYDVQQLDApUZWNobm9sb2d5MQ4wDAYDVQQDDAVnZW5p
ZTElMCMGCSqGSIb3DQEJARYWemFpZC5hYmR1bGxhQGZhbm9vcy5pcTCBnzANBgkq
hkiG9w0BAQEFAAOBjQAwgYkCgYEAxtHC3jlOoVsZQ9I8Wgd+8NkEaRfc01lRIAzj
eWuUO/xNY54EUDHecRxwolHtoGkkZAp4kKR3uUVdAo9uXmNT2CeeZ5lfRoa8WmgG
Yd6PExsdc4ZeKcw20XrKtQICqmU6rGlLERMFX97ejQh3eUmqwbzNaXiURxcKGdeK
2C/NxoUCAwEAATANBgkqhkiG9w0BAQUFAAOBgQAYFw6dy/z74ahnA5QpuJOjjTrt
AJaL9iuL1xS6TFfzne9UMIT+fFJsgSy6xx3TGkvbZJTnCzJN7I2uqmAhMoGFy7Ea
svCCkj8U5qkjqgwwi7IbwQ7fXxTVdPqUlkIdNB2YLDELksxCX+ZzGQPiFB8WrSoO
OBvB+hhPFpgp3aO2AQ==
-----END CERTIFICATE-----
config/httpscert-sample.key
-15
View File
@@ -1,15 +0,0 @@
-----BEGIN RSA PRIVATE KEY-----
MIICXgIBAAKBgQDG0cLeOU6hWxlD0jxaB37w2QRpF9zTWVEgDON5a5Q7/E1jngRQ
Md5xHHCiUe2gaSRkCniQpHe5RV0Cj25eY1PYJ55nmV9GhrxaaAZh3o8TGx1zhl4p
zDbResq1AgKqZTqsaUsREwVf3t6NCHd5SarBvM1peJRHFwoZ14rYL83GhQIDAQAB
AoGAR4NpuW8ZhlZzNv7m8Gcj1NPqjzajGm5VDNc19xWoAw+UcpGC8amOXlcheHFJ
rNpE2asehMBzw/NcKjCXPWIy5LnGmttJMN2EbT2h/+uxBIM5IUc+joF744xF4C0t
5zrtqQYJxC7sBLX2vf7nY6iuovUM7fHEwACyV3Rpw3zdZf0CQQDiqW+ugLKcfS/c
6hkGTxeYWF4gUJTf3ke3au8nyMmUhyBwOuP4kpO72mK8bMLtjZRk3g6YChA4qH41
nfXUXIh7AkEA4I2/zj4GDjjspDR+/tX0iUHQk1Rm2CRmRXXFJV1f688OLODCcOFK
Ms/+qnbe41wXo1NUxCqVrnPwEh3MVl88/wJBANUTpXqtX7s+8ETECTxmE+9Se0Hz
ntCJXG/v5d1Ho0Cueu8mXXMrEBiTpygtMl4xL2RXX+7L3w30T5lF4nh9eIsCQQDY
DmsbdL/81TaTc9MxPQlxOzsFVrMxiixA7uTGY8ZULD/2XDeZxprOsyPUPlWMquIv
ipFHwbWNSg4WUEd4X2BfAkEAytze+0mCd0WCsKbKwfTQnmJ3bK7JQywtC03llWyA
Om3Vl61Syaa7M2hl5rLvitZMvU1Wd0s8raOXN8RUuRXZBw==
-----END RSA PRIVATE KEY-----
lib/cluster.coffee
+45
View File
@@ -0,0 +1,45 @@
###
# Copyright 2013, 2014 Zaid Abdulla
#
# GenieACS is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# GenieACS is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
###
config = require './config'
cluster = require 'cluster'
util = require 'util'
cluster.on('listening', (worker, address) ->
util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
)
cluster.on('exit', (worker, code, signal) ->
util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
cluster.fork()
)
start = (service) ->
workerProcesses = config["#{service.toUpperCase()}_WORKER_PROCESSES"]
cluster.setupMaster({
exec : 'lib/server',
args : [service]
})
for [0 ... workerProcesses]
cluster.fork()
exports.start = start
lib/config.coffee
+86 -34
View File
@@ -15,43 +15,95 @@
# along with GenieACS. If not, see <http://www.gnu.org/licenses/>. # along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
### ###
exports.MONGODB_CONNECTION_URL = 'mongodb://localhost/genieacs' path = require 'path'
exports.REDIS_DB = 0 common = require './common'
exports.ACS_PORT = 7547 # CWMP port as assigned by IANA
exports.ACS_HTTPS_PORT = 7548
exports.ACS_INTERFACE = '0.0.0.0' defaults = {
exports.ACS_HTTPS_INTERFACE = '0.0.0.0' CONFIG_DIR : './config',
exports.API_PORT = 7557
exports.API_INTERFACE = '0.0.0.0' MONGODB_CONNECTION_URL : 'mongodb://127.0.0.1/genieacs',
exports.FILES_PORT = 7567 REDIS_PORT : 6379,
exports.FILES_INTERFACE = '0.0.0.0' REDIS_HOST : '127.0.0.1',
exports.FILES_IP = '172.240.21.2' # Used when sending download requests to devices REDIS_DB : 0,
exports.CACHE_DURATION = 60 # in seconds
exports.PRESETS_CACHE_DURATION = 86400 CWMP_WORKER_PROCESSES : 4,
exports.PRESETS_TIME_PADDING = 1 CWMP_PORT : 7547, # CWMP port as assigned by IANA
exports.MONGODB_SOCKET = '/tmp/mongodb-27017.sock' CWMP_INTERFACE : '0.0.0.0',
exports.REDIS_SOCKET = '/tmp/redis.sock' CWMP_SSL : false,
exports.WORKER_RESPAWN_TIME = 60000 NBI_WORKER_PROCESSES : 2,
exports.LOG_INFORMS = false NBI_PORT : 7557,
exports.DEBUG_DEVICES = {} # {'202BC1-BM632w-8KA8WA1151100043' : true} NBI_INTERFACE : '0.0.0.0',
exports.DEVICE_ONLINE_THRESHOLD = 4000 NBI_SSL : false,
exports.RETRY_DELAY = 300 FS_WORKER_PROCESSES : 2,
exports.IGNORE_XML_NAMESPACES = false # Traverse XML using element's local name only FS_PORT : 7567,
exports.LIBXMLJS_OPTIONS = {} # Options passed directly to libxmljs FS_INTERFACE : '0.0.0.0',
exports.SESSION_TIMEOUT = 30 FS_SSL : false,
exports.GET_PARAMETER_NAMES_DEPTH_THRESHOLD = 0 FILES_IP : '192.168.0.1', # Used when sending download requests to devices
exports.TASK_PARAMETERS_BATCH_SIZE = 32
CACHE_DURATION : 60, # in seconds
PRESETS_CACHE_DURATION : 86400,
PRESETS_TIME_PADDING : 1,
LOG_INFORMS : true,
DEBUG : false,
DEVICE_ONLINE_THRESHOLD : 4000,
RETRY_DELAY : 300,
IGNORE_XML_NAMESPACES : false, # Traverse XML using element's local name only
SESSION_TIMEOUT : 30,
GET_PARAMETER_NAMES_DEPTH_THRESHOLD : 0,
TASK_PARAMETERS_BATCH_SIZE : 32
}
# Command line arguments
exports.argv = []
argv = process.argv[2..]
while argv.length
arg = argv.shift()
if arg[0] == '-'
v = argv.shift()
exports.argv[arg] = v
n = arg[2..].toUpperCase().replace(/-/g, '_')
if defaults[n]?
exports[n] ?= common.matchType(defaults[n], v)
# Save as environmnet variables to pass on to any child processes
process.env["GENIEACS_#{n}"] = v
else
exports.argv.push(arg)
# Environment variable
for k, v of process.env
k = k[9..] # remove "GENIEACS_" prefix
if defaults[k]?
exports[k] ?= common.matchType(defaults[k], v)
# Find config dir
if exports.argv['--config-dir']?
exports.CONFIG_DIR = exports.argv['--config-dir']
else if process.env['GENIEACS_CONFIG_DIR']?
exports.CONFIG_DIR = process.env['GENIEACS_CONFIG_DIR']
else
exports.CONFIG_DIR = defaults.CONFIG_DIR
# Configuration file
for k, v of require(path.resolve(exports.CONFIG_DIR, 'config'))
if defaults[k]?
exports[k] ?= common.matchType(defaults[k], v)
# Defaults
for k, v of defaults
exports[k] ?= defaults[k]
# load configuration
c = require('../config/config')
for k, v of c
exports[k] = v
# load parameter configurations # load parameter configurations
exports.PARAMETERS = require('../config/parameters') exports.PARAMETERS = require(path.resolve(exports.CONFIG_DIR, 'parameters'))
# load authentication scripts # load authentication scripts
exports.auth = require('../config/auth') exports.auth = require(path.resolve(exports.CONFIG_DIR, 'auth'))
exports.CUSTOM_COMMANDS = require('../config/custom_commands')
exports.CUSTOM_COMMANDS = require(path.resolve(exports.CONFIG_DIR, 'custom_commands'))
lib/cwmp.coffee
+4 -41
View File
@@ -41,14 +41,11 @@
config = require './config' config = require './config'
common = require './common' common = require './common'
util = require 'util' util = require 'util'
http = require 'http'
https = require 'https'
soap = require './soap' soap = require './soap'
tasks = require './tasks' tasks = require './tasks'
db = require './db' db = require './db'
presets = require './presets' presets = require './presets'
mongodb = require 'mongodb' mongodb = require 'mongodb'
fs = require 'fs'
apiFunctions = require './api-functions' apiFunctions = require './api-functions'
customCommands = require './custom-commands' customCommands = require './custom-commands'
crypto = require 'crypto' crypto = require 'crypto'
@@ -61,9 +58,9 @@ holdUntil = Date.now()
writeResponse = (currentRequest, res) -> writeResponse = (currentRequest, res) ->
if config.DEBUG_DEVICES[currentRequest.session.deviceId] if config.DEBUG
dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n" dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n"
fs = require('fs').appendFile("../debug/#{currentRequest.session.deviceId}.dump", dump, (err) -> require('fs').appendFile("../debug/#{currentRequest.session.deviceId}.dump", dump, (err) ->
throw err if err throw err if err
) )
@@ -454,7 +451,7 @@ listener = (httpRequest, httpResponse) ->
session : session session : session
} }
if config.DEBUG_DEVICES[currentRequest.session.deviceId] if config.DEBUG
dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n" dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n"
require('fs').appendFile("../debug/#{currentRequest.session.deviceId}.dump", dump, (err) -> require('fs').appendFile("../debug/#{currentRequest.session.deviceId}.dump", dump, (err) ->
throw err if err throw err if err
@@ -562,39 +559,5 @@ listener = (httpRequest, httpResponse) ->
) )
) )
cluster = require 'cluster'
numCPUs = require('os').cpus().length
if cluster.isMaster exports.listener = listener
db.connect((err) ->
throw err if err
db.redisClient.flushdb()
db.disconnect()
cluster.on('listening', (worker, address) ->
util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
)
cluster.on('exit', (worker, code, signal) ->
util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
setTimeout(()->
cluster.fork()
, config.WORKER_RESPAWN_TIME)
)
for [1 .. numCPUs]
cluster.fork()
)
else
options = {
key: fs.readFileSync('../config/httpscert.key'),
cert: fs.readFileSync('../config/httpscert.crt')
}
httpServer = http.createServer(listener)
httpsServer = https.createServer(options, listener)
db.connect((err) ->
throw err if err
httpServer.listen(config.ACS_PORT, config.ACS_INTERFACE)
httpsServer.listen(config.ACS_HTTPS_PORT, config.ACS_HTTPS_INTERFACE)
)
lib/db.coffee
+1 -1
View File
@@ -70,7 +70,7 @@ connect = (callback) ->
return callback(err) return callback(err)
) )
exports.redisClient = redisClient = redis.createClient(config.REDIS_SOCKET) exports.redisClient = redisClient = redis.createClient(config.REDIS_PORT, config.REDIS_HOST)
redisClient.select(config.REDIS_DB, (err) -> redisClient.select(config.REDIS_DB, (err) ->
if --callbackCounter == 0 or err if --callbackCounter == 0 or err
callbackCounter = 0 callbackCounter = 0
lib/fs.coffee
+19 -43
View File
@@ -21,53 +21,29 @@
### ###
config = require './config' config = require './config'
util = require 'util'
http = require 'http'
url = require 'url' url = require 'url'
mongodb = require 'mongodb' mongodb = require 'mongodb'
querystring = require 'querystring' querystring = require 'querystring'
db = require './db'
db = null
cluster = require 'cluster' listener = (request, response) ->
numCPUs = require('os').cpus().length urlParts = url.parse(request.url, true)
if request.method == 'GET'
filename = querystring.unescape(urlParts.pathname.substring(1))
gs = new mongodb.GridStore(db.mongoDb, filename, 'r', {})
gs.open((err, gs) ->
if err
response.writeHead(404)
response.end()
return
stream = gs.stream(true)
response.writeHead(200, {'Content-Type' : 'application/octet-stream', 'Content-Length' : gs.length})
stream.pipe(response)
)
else
response.writeHead(405, {'Allow': 'GET'})
response.end('405 Method Not Allowed')
if cluster.isMaster
cluster.on('listening', (worker, address) ->
util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}")
)
cluster.on('exit', (worker, code, signal) -> exports.listener = listener
util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})")
setTimeout(()->
cluster.fork()
, config.WORKER_RESPAWN_TIME)
)
for i in [1 .. numCPUs]
cluster.fork()
else
server = http.createServer((request, response) ->
urlParts = url.parse(request.url, true)
if request.method == 'GET'
filename = querystring.unescape(urlParts.pathname.substring(1))
gs = new mongodb.GridStore(db, filename, 'r', {})
gs.open((err, gs) ->
if err
response.writeHead(404)
response.end()
return
stream = gs.stream(true)
response.writeHead(200, {'Content-Type' : 'application/octet-stream', 'Content-Length' : gs.length})
stream.pipe(response)
)
else
response.writeHead(405, {'Allow': 'GET'})
response.end('405 Method Not Allowed')
)
mongodb.MongoClient.connect("mongodb://#{config.MONGODB_SOCKET}/#{config.DATABASE_NAME}", config.MONGODB_OPTIONS, (err, _db) ->
throw err if err
db = _db
server.listen config.FILES_PORT, config.FILES_INTERFACE
)
lib/nbi.coffee
+288 -310
View File
@@ -39,8 +39,6 @@
### ###
config = require './config' config = require './config'
util = require 'util'
http = require 'http'
url = require 'url' url = require 'url'
common = require './common' common = require './common'
db = require './db' db = require './db'
@@ -62,343 +60,323 @@ QUERY_REGEX = /^\/([a-zA-Z0-9_]+s)\/?$/
DEVICE_PRESET_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/preset\/?$/ DEVICE_PRESET_REGEX = /^\/devices\/([a-zA-Z0-9\-\_\%]+)\/preset\/?$/
cluster = require 'cluster' listener = (request, response) ->
numCPUs = require('os').cpus().length chunks = []
bytes = 0
if cluster.isMaster request.addListener 'data', (chunk) ->
cluster.on('listening', (worker, address) -> chunks.push(chunk)
util.log("Worker #{worker.process.pid} listening to #{address.address}:#{address.port}") bytes += chunk.length
)
cluster.on('exit', (worker, code, signal) -> request.getBody = () ->
util.log("Worker #{worker.process.pid} died (#{worker.process.exitCode})") # Write all chunks into a Buffer
setTimeout(()-> body = new Buffer(bytes)
cluster.fork() offset = 0
, config.WORKER_RESPAWN_TIME) chunks.forEach((chunk) ->
) chunk.copy(body, offset, 0, chunk.length)
offset += chunk.length
)
return body
for [1 .. numCPUs] request.addListener 'end', () ->
cluster.fork() body = request.getBody()
else urlParts = url.parse(request.url, true)
server = http.createServer (request, response) -> if PRESETS_REGEX.test(urlParts.pathname)
chunks = [] presetName = querystring.unescape(PRESETS_REGEX.exec(urlParts.pathname)[1])
bytes = 0 if request.method == 'PUT'
preset = JSON.parse(body)
preset._id = presetName
request.addListener 'data', (chunk) -> db.presetsCollection.save(preset, (err) ->
chunks.push(chunk) db.redisClient.del('presets', 'presets_hash', (err) ->
bytes += chunk.length throw err if err
request.getBody = () ->
# Write all chunks into a Buffer
body = new Buffer(bytes)
offset = 0
chunks.forEach((chunk) ->
chunk.copy(body, offset, 0, chunk.length)
offset += chunk.length
)
return body
request.addListener 'end', () ->
body = request.getBody()
urlParts = url.parse(request.url, true)
if PRESETS_REGEX.test(urlParts.pathname)
presetName = querystring.unescape(PRESETS_REGEX.exec(urlParts.pathname)[1])
if request.method == 'PUT'
preset = JSON.parse(body)
preset._id = presetName
db.presetsCollection.save(preset, (err) ->
db.redisClient.del('presets', 'presets_hash', (err) ->
throw err if err
)
if err
response.writeHead(500)
response.end(err)
return
response.writeHead(200)
response.end()
) )
else if request.method == 'DELETE' if err
db.presetsCollection.remove({'_id' : presetName}, (err, removedCount) -> response.writeHead(500)
db.redisClient.del('presets', 'presets_hash', (err) -> response.end(err)
throw err if err return
) response.writeHead(200)
if err response.end()
response.writeHead(500) )
response.end(err) else if request.method == 'DELETE'
return db.presetsCollection.remove({'_id' : presetName}, (err, removedCount) ->
response.writeHead(200) db.redisClient.del('presets', 'presets_hash', (err) ->
response.end() throw err if err
) )
else if err
response.writeHead 405, {'Allow': 'PUT, DELETE'} response.writeHead(500)
response.end('405 Method Not Allowed') response.end(err)
else if OBJECTS_REGEX.test(urlParts.pathname) return
objectName = querystring.unescape(OBJECTS_REGEX.exec(urlParts.pathname)[1]) response.writeHead(200)
if request.method == 'PUT' response.end()
object = JSON.parse(body) )
object._id = objectName else
response.writeHead 405, {'Allow': 'PUT, DELETE'}
response.end('405 Method Not Allowed')
else if OBJECTS_REGEX.test(urlParts.pathname)
objectName = querystring.unescape(OBJECTS_REGEX.exec(urlParts.pathname)[1])
if request.method == 'PUT'
object = JSON.parse(body)
object._id = objectName
db.objectsCollection.save(object, (err) -> db.objectsCollection.save(object, (err) ->
db.redisClient.del('objects', 'presets_hash', (err) -> db.redisClient.del('objects', 'presets_hash', (err) ->
throw err if err throw err if err
)
if err
response.writeHead(500)
response.end(err)
return
response.writeHead(200)
response.end()
) )
else if request.method == 'DELETE' if err
db.objectsCollection.remove({'_id' : objectName}, (err, removedCount) -> response.writeHead(500)
db.redisClient.del('objects', 'presets_hash', (err) -> response.end(err)
throw err if err return
) response.writeHead(200)
if err response.end()
response.writeHead(500) )
response.end(err) else if request.method == 'DELETE'
return db.objectsCollection.remove({'_id' : objectName}, (err, removedCount) ->
response.writeHead(200) db.redisClient.del('objects', 'presets_hash', (err) ->
response.end() throw err if err
) )
else if err
response.writeHead 405, {'Allow': 'PUT, DELETE'} response.writeHead(500)
response.end('405 Method Not Allowed') response.end(err)
else if TAGS_REGEX.test(urlParts.pathname) return
r = TAGS_REGEX.exec(urlParts.pathname) response.writeHead(200)
deviceId = querystring.unescape(r[1]) response.end()
tag = querystring.unescape(r[2]) )
if request.method == 'POST' else
db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) -> response.writeHead 405, {'Allow': 'PUT, DELETE'}
db.redisClient.del("#{deviceId}_presets_hash", (err) -> response.end('405 Method Not Allowed')
throw err if err else if TAGS_REGEX.test(urlParts.pathname)
) r = TAGS_REGEX.exec(urlParts.pathname)
if err deviceId = querystring.unescape(r[1])
response.writeHead(500) tag = querystring.unescape(r[2])
response.end(err) if request.method == 'POST'
return db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) ->
response.writeHead(200) db.redisClient.del("#{deviceId}_presets_hash", (err) ->
response.end() throw err if err
) )
else if request.method == 'DELETE' if err
db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) -> response.writeHead(500)
db.redisClient.del("#{deviceId}_presets_hash", (err) -> response.end(err)
throw err if err return
) response.writeHead(200)
if err response.end()
response.writeHead(500) )
response.end(err) else if request.method == 'DELETE'
return db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) ->
response.writeHead(200) db.redisClient.del("#{deviceId}_presets_hash", (err) ->
response.end() throw err if err
) )
else if err
response.writeHead 405, {'Allow': 'POST, DELETE'} response.writeHead(500)
response.end('405 Method Not Allowed') response.end(err)
else if DEVICE_TASKS_REGEX.test(urlParts.pathname) return
if request.method == 'POST' response.writeHead(200)
deviceId = querystring.unescape(DEVICE_TASKS_REGEX.exec(urlParts.pathname)[1]) response.end()
if body.length )
task = JSON.parse(body) else
task.device = deviceId response.writeHead 405, {'Allow': 'POST, DELETE'}
db.getAliases((aliases) -> response.end('405 Method Not Allowed')
apiFunctions.insertTasks(task, aliases, (err) -> else if DEVICE_TASKS_REGEX.test(urlParts.pathname)
db.redisClient.del("#{deviceId}_presets_hash", (err) -> if request.method == 'POST'
throw err if err deviceId = querystring.unescape(DEVICE_TASKS_REGEX.exec(urlParts.pathname)[1])
) if body.length
if err task = JSON.parse(body)
response.writeHead(500) task.device = deviceId
response.end(err) db.getAliases((aliases) ->
return apiFunctions.insertTasks(task, aliases, (err) ->
db.redisClient.del("#{deviceId}_presets_hash", (err) ->
if urlParts.query.connection_request? throw err if err
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
apiFunctions.connectionRequest(deviceId, (err) ->
if err
response.writeHead 504
response.end()
return
response.writeHead 200
response.end()
)
else
response.writeHead(400)
response.end()
else
response.writeHead 405, {'Allow': 'POST'}
response.end('405 Method Not Allowed')
else if DEVICE_PRESET_REGEX.test(urlParts.pathname)
deviceId = querystring.unescape(DEVICE_PRESET_REGEX.exec(urlParts.pathname)[1])
if request.method is 'GET'
db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) ->
presets.getDevicePreset(deviceId, allPresets, allObjects, allAliases, (devicePreset) ->
response.writeHead(200, {'Content-Type' : 'application/json'})
response.end(JSON.stringify(devicePreset))
)
)
else
response.writeHead 405, {'Allow': 'GET'}
response.end('405 Method Not Allowed')
else if TASKS_REGEX.test(urlParts.pathname)
r = TASKS_REGEX.exec(urlParts.pathname)
taskId = mongodb.ObjectID(querystring.unescape(r[1]))
action = try querystring.unescape(r[2])
if not action? or action is '/'
if request.method == 'DELETE'
db.tasksCollection.remove({'_id' : taskId}, (err, removedCount) ->
if err if err
response.writeHead(500) response.writeHead(500)
response.end(err) response.end(err)
return return
response.writeHead(200)
response.end()
)
else
response.writeHead 405, {'Allow': 'PUT DELETE'}
response.end('405 Method Not Allowed')
else if action is '/retry'
if request.method == 'POST'
db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $set : {timestamp : new Date()}}, (err, count) ->
# TODO need to invalidate presets hash for the device
response.writeHead(200)
response.end()
)
else
response.writeHead 405, {'Allow': 'POST'}
response.end('405 Method Not Allowed')
else
response.writeHead(404)
response.end()
else if FILES_REGEX.test(urlParts.pathname)
filename = querystring.unescape(FILES_REGEX.exec(urlParts.pathname)[1])
if request.method == 'PUT'
metadata = {
fileType : request.headers.filetype,
manufacturer : request.headers.manufacturer,
productClass : request.headers.productclass,
version : request.headers.version,
}
gs = new mongodb.GridStore(db.mongoDb, filename, 'w', {metadata : metadata}) if urlParts.query.connection_request?
gs.open((err, gs) -> apiFunctions.connectionRequest(deviceId, (err) ->
gs.write(body, (err, res) -> if err
throw err if err response.writeHead(202)
gs.close((err) -> response.end()
throw err if err else
response.writeHead(201) 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() response.end()
)
) )
) )
else if request.method == 'DELETE' else if urlParts.query.connection_request?
mongodb.GridStore.unlink(db.mongoDb, filename, (err) -> # no task, send connection request only
apiFunctions.connectionRequest(deviceId, (err) ->
if err
response.writeHead 504
response.end()
return
response.writeHead 200
response.end()
)
else
response.writeHead(400)
response.end()
else
response.writeHead 405, {'Allow': 'POST'}
response.end('405 Method Not Allowed')
else if DEVICE_PRESET_REGEX.test(urlParts.pathname)
deviceId = querystring.unescape(DEVICE_PRESET_REGEX.exec(urlParts.pathname)[1])
if request.method is 'GET'
db.getPresetsObjectsAliases((allPresets, allObjects, allAliases) ->
presets.getDevicePreset(deviceId, allPresets, allObjects, allAliases, (devicePreset) ->
response.writeHead(200, {'Content-Type' : 'application/json'})
response.end(JSON.stringify(devicePreset))
)
)
else
response.writeHead 405, {'Allow': 'GET'}
response.end('405 Method Not Allowed')
else if TASKS_REGEX.test(urlParts.pathname)
r = TASKS_REGEX.exec(urlParts.pathname)
taskId = mongodb.ObjectID(querystring.unescape(r[1]))
action = try querystring.unescape(r[2])
if not action? or action is '/'
if request.method == 'DELETE'
db.tasksCollection.remove({'_id' : taskId}, (err, removedCount) ->
if err
response.writeHead(500)
response.end(err)
return
response.writeHead(200) response.writeHead(200)
response.end() response.end()
) )
else else
response.writeHead 405, {'Allow': 'PUT, DELETE'} response.writeHead 405, {'Allow': 'PUT DELETE'}
response.end('405 Method Not Allowed') response.end('405 Method Not Allowed')
else if PING_REGEX.test(urlParts.pathname) else if action is '/retry'
host = querystring.unescape(PING_REGEX.exec(urlParts.pathname)[1]) if request.method == 'POST'
require('child_process').exec("ping -w 1 -i 0.2 -c 3 #{host}", (err, stdout, stderr) -> db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $set : {timestamp : new Date()}}, (err, count) ->
if err? # TODO need to invalidate presets hash for the device
response.writeHead(404, {'Cache-Control' : 'no-cache'}) response.writeHead(200)
response.end() response.end()
return
response.writeHead(200, {'Content-Type' : 'text/plain', 'Cache-Control' : 'no-cache'})
response.end(stdout)
)
else if QUERY_REGEX.test(urlParts.pathname)
collectionName = QUERY_REGEX.exec(urlParts.pathname)[1]
if request.method not in ['GET', 'HEAD']
response.writeHead 405, {'Allow' : 'GET, HEAD'}
response.end('405 Method Not Allowed')
return
collection = db["#{collectionName}Collection"]
if not collection?
response.writeHead 404
response.end('404 Not Found')
return
func = (aliases) ->
if urlParts.query.query?
q = JSON.parse(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 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(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 else
func({}) response.writeHead 405, {'Allow': 'POST'}
response.end('405 Method Not Allowed')
else else
response.writeHead(404)
response.end()
else if FILES_REGEX.test(urlParts.pathname)
filename = querystring.unescape(FILES_REGEX.exec(urlParts.pathname)[1])
if request.method == 'PUT'
metadata = {
fileType : request.headers.filetype,
manufacturer : request.headers.manufacturer,
productClass : request.headers.productclass,
version : request.headers.version,
}
gs = new mongodb.GridStore(db.mongoDb, filename, 'w', {metadata : metadata})
gs.open((err, gs) ->
gs.write(body, (err, res) ->
throw err if err
gs.close((err) ->
throw err if err
response.writeHead(201)
response.end()
)
)
)
else if request.method == 'DELETE'
mongodb.GridStore.unlink(db.mongoDb, filename, (err) ->
response.writeHead(200)
response.end()
)
else
response.writeHead 405, {'Allow': 'PUT, DELETE'}
response.end('405 Method Not Allowed')
else if PING_REGEX.test(urlParts.pathname)
host = querystring.unescape(PING_REGEX.exec(urlParts.pathname)[1])
require('child_process').exec("ping -w 1 -i 0.2 -c 3 #{host}", (err, stdout, stderr) ->
if err?
response.writeHead(404, {'Cache-Control' : 'no-cache'})
response.end()
return
response.writeHead(200, {'Content-Type' : 'text/plain', 'Cache-Control' : 'no-cache'})
response.end(stdout)
)
else if QUERY_REGEX.test(urlParts.pathname)
collectionName = QUERY_REGEX.exec(urlParts.pathname)[1]
if request.method not in ['GET', 'HEAD']
response.writeHead 405, {'Allow' : 'GET, HEAD'}
response.end('405 Method Not Allowed')
return
collection = db["#{collectionName}Collection"]
if not collection?
response.writeHead 404 response.writeHead 404
response.end('404 Not Found') response.end('404 Not Found')
return
db.connect((err) -> func = (aliases) ->
throw err if err if urlParts.query.query?
server.listen config.API_PORT, config.API_INTERFACE q = JSON.parse(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 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(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')
exports.listener = listener
lib/server.coffee
+47
View File
@@ -0,0 +1,47 @@
###
# Copyright 2013, 2014 Zaid Abdulla
#
# GenieACS is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# GenieACS is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
###
config = require './config'
db = require './db'
service = config.argv[0]
if not service?
throw new Error('Missing argument cwmp, fs, or nbi')
networkInterface = config["#{service.toUpperCase()}_INTERFACE"]
port = config["#{service.toUpperCase()}_PORT"]
useHttps = config["#{service.toUpperCase()}_SSL"]
listener = require("./#{service}").listener
if useHttps
path = require 'path'
fs = require 'fs'
httpsKey = path.resolve(config.CONFIG_DIR, "#{service}.key")
httpsCert = path.resolve(config.CONFIG_DIR, "#{service}.crt")
options = {
key: fs.readFileSync(httpsKey),
cert: fs.readFileSync(httpsCert)
}
server = require('https').createServer(options, listener)
else
server = require('http').createServer(listener)
db.connect((err) ->
throw err if err
server.listen(port, networkInterface)
)
package.json
+1 -1
View File
@@ -34,7 +34,7 @@
"scripts": { "scripts": {
"prepublish": "npm run compile", "prepublish": "npm run compile",
"compile": "coffee -cb lib/*.coffee test/*.coffee tools/*.coffee", "compile": "coffee -cb lib/*.coffee test/*.coffee tools/*.coffee",
"configure": "cp ./config/config-sample.json ./config/config.json; cp ./config/parameters-sample.json ./config/parameters.json; cp ./config/custom_commands-sample.json ./config/custom_commands.json; cp ./config/auth-sample.js ./config/auth.js; cp ./config/httpscert-sample.key ./config/httpscert.key; cp ./config/httpscert-sample.crt ./config/httpscert.crt", "configure": "cp ./config/config-sample.json ./config/config.json; cp ./config/parameters-sample.json ./config/parameters.json; cp ./config/custom_commands-sample.json ./config/custom_commands.json; cp ./config/auth-sample.js ./config/auth.js",
"install": "npm run configure", "install": "npm run configure",
"test": "node test" "test": "node test"
} }