Skip to content

V
vernemq
Commit cefb2042 by zCaesar
Options

modify redis statement

parent a23d9730
Showing with 96 additions and 257 deletions
authhook/auth_on_publish.js
var config = require('config'); var config = require('config');
// https://github.com/isaacs/node-lru-cache // https://github.com/isaacs/node-lru-cache
var on_message_redis = require('./redis/on_message').on_message_redis
var LRU = require("lru-cache"), var LRU = require("lru-cache"),
cache = LRU({ cache = LRU({
max: 500, max: 500,
...@@ -17,65 +19,63 @@ module.exports = function (options = {}) { ...@@ -17,65 +19,63 @@ module.exports = function (options = {}) {
var topic = req.body.topic; var topic = req.body.topic;
// save redis var authstatus;
require('./redis/on_message').on_message_redis(req.body.client_id, req.body.payload, topic).then(status => {
if (status) {
var authstatus;
var cachekey = 'pub:' + req.body.client_id + ':' + req.body.username + ':' + topic; var cachekey = 'pub:' + req.body.client_id + ':' + req.body.username + ':' + topic;
var authstatus = cache.get(cachekey); var authstatus = cache.get(cachekey);
if (typeof (authstatus) === 'undefined') { if (typeof (authstatus) === 'undefined') {
cache.set(cachekey, true); // cache missed cache.set(cachekey, true); // cache missed
} }
var response var response
var decoded = require('jwt-verify').verify(req.body.username).res var decoded = require('jwt-verify').verify(req.body.username).res
if (decoded) {
if (decoded.role === 'realtimedb') response = { 'result': 'ok' }
else response = { result: { error: 'not allowed' } }
res.send(response);
next();
}
else {
var GGID = require('./utils/getGroupID');
if (decoded) { var output = {};
if (decoded.role === 'realtimedb') response = { 'result': 'ok' } GGID.getGroupID(req.body.username, req.body.client_id, function (group) {
else response = { result: {error: 'not allowed'} } if (group) {
var _ftopic = require('./utils/router').rewriteTopic(topic, 'pub', group, req.body.client_id, output); // get topic where concat with groupID
response = {
'result': 'ok',
'modifiers': {
'topic': _ftopic,
'qos': 0,
'retain': false
}
}
if (options.debug) {
console.log(response);
}
if (output.verb == 'get' || output.verb == 'read') {
response.modifiers.payload = Buffer.from(req.body.client_id).toString('base64');
}
// save on redis before send response
on_message_redis(req.body.client_id, req.body.payload, topic)
res.send(response); res.send(response);
next(); next();
} }
else { else {
var GGID = require('./utils/getGroupID'); res.send({ result: { error: 'not allowed' } });
next();
var output = {};
GGID.getGroupID(req.body.username, req.body.client_id, function (group) {
if (group) {
var _ftopic = require('./utils/router').rewriteTopic(topic, 'pub', group, req.body.client_id, output); // get topic where concat with groupID
response = {
'result': 'ok',
'modifiers': {
'topic': _ftopic,
'qos': 0,
'retain': false
}
}
if (options.debug) {
console.log(response);
}
if (output.verb == 'get' || output.verb == 'read') {
response.modifiers.payload = Buffer.from(req.body.client_id).toString('base64');
}
res.send(response);
next();
}
else {
res.send({ result: {error: 'not allowed'} });
next();
}
});
} }
} });
}) }
} }
else { else {
res.send({ result: {error: 'not allowed'} }); res.send({ result: { error: 'not allowed' } });
next(); next();
} }
} }
......
authhook/auth_on_register.js
...@@ -3,7 +3,7 @@ var validator = require('./validator'); ...@@ -3,7 +3,7 @@ var validator = require('./validator');
var config = require('config'); var config = require('config');
// var seneca = require('seneca')({log: 'silent'}).client({ port: config.get('device_registry_port'), host: config.get('device_registry_host') }); // var seneca = require('seneca')({log: 'silent'}).client({ port: config.get('device_registry_port'), host: config.get('device_registry_host') });
var seneca = require('seneca')({ log: 'silent' }).client({ port: config.get('token_registry_port'), host: config.get('token_registry_host') }); var seneca = require('seneca')({ log: 'silent' }).client({ port: config.get('token_registry_port'), host: config.get('token_registry_host') });
let on_register_redis = require('./redis/on_register').on_register_redis var on_register_redis = require('./redis/on_register').on_register_redis
// https://github.com/isaacs/node-lru-cache // https://github.com/isaacs/node-lru-cache
var LRU = require("lru-cache"), var LRU = require("lru-cache"),
...@@ -45,45 +45,6 @@ function authCheck(client_id, token, password, callback) { ...@@ -45,45 +45,6 @@ function authCheck(client_id, token, password, callback) {
callback(false); callback(false);
} }
}); });
// seneca.act('cmd:getAccessTokenInfo, tokencode:'+token, function(err,res) {
// if (!err && res) {
// var token_profile = (res&&res.result&&res.result[0])?res.result[0]:{};
// var mqttauth = {
// clientid : client_id,
// token : token,
// password : password
// };
// callback( validator.auth_connect(mqttauth, token_profile) );
// }
// else {
// callback(false);
// }
// });
// authclient.act({ role: 'auth', cmd: 'token', action: 'info', token: token }, function (err, res) { // auth client device by query from db
// if (debug) {
// console.log("res ------>\n");
// console.log(res);
// }
// if (res && res.data) {
// try {
// var jdata = JSON.parse(res.data);
// if (jdata && jdata.code == 200) {
// callback(true);
// }
// else callback(false);
// } catch (e) {
// callback(false);
// }
// callback(true);
// }
// else {
// callback(false);
// }
// });
} }
} }
...@@ -101,85 +62,15 @@ module.exports = function (options = {}) { ...@@ -101,85 +62,15 @@ module.exports = function (options = {}) {
} }
var cachekey = 'auth:' + req.body.client_id + ':' + req.body.username + ':' + req.body.password; var cachekey = 'auth:' + req.body.client_id + ':' + req.body.username + ':' + req.body.password;
var authstatus = cache.get(cachekey); var authstatus = cache.get(cachekey);
<<<<<<< HEAD
// if (typeof (authstatus) == 'undefined') {
// // cache missed
// authCheck(req.body.client_id, req.body.username, req.body.password, function (result) {
// cache.set(cachekey, result);
// if (result) {
// require('./redis/on_register').on_register_redis(req.body.client_id).then(status => {
// console.log('redis:status: ' + status)
// if (status) {
// res.send({
// "result": "ok"
// });
// } else {
// res.send({
// result: { error: 'not allowed' }
// });
// }
// })
// } else {
// res.send({
// result: {error: 'not allowed'}
// });
// }
// // const util = require('util')
// // console.log(util.inspect(status, false, null, true))
// });
// }
// else {
// if (authstatus) {
// require('./redis/on_register').on_register_redis(req.body.client_id).then(status => {
// console.log('redis:status: ' + status)
// if (status) {
// res.send({
// "result": "ok"
// });
// } else {
// res.send({
// result: {error: 'not allowed'}
// });
// }
// })
// } else {
// res.send({
// result: {error: 'not allowed'}
// });
// }
// }
on_register_redis(req.body.client_id).then(status => {
console.log('redis:status: ' + status)
if (status) {
res.send({
"result": "ok"
});
} else {
res.send({
result: { error: 'not allowed' }
=======
if (typeof (authstatus) == 'undefined') { if (typeof (authstatus) == 'undefined') {
// cache missed // cache missed
authCheck(req.body.client_id, req.body.username, req.body.password, function (result) { authCheck(req.body.client_id, req.body.username, req.body.password, function (result) {
cache.set(cachekey, result); cache.set(cachekey, result);
if (result) { if (result) {
require('./redis/on_register').on_register_redis(req.body.client_id).then(status => { on_register_redis(req.body.client_id)
console.log('redis:status: ' + status) res.send({ result: 'ok' });
if (status) {
res.send({
result: 'ok'
});
} else {
res.send({
result: {error: 'not allowed'}
});
}
})
} else { } else {
res.send({ res.send({ result: { error: 'not allowed' } });
result: {error: 'not allowed'}
});
} }
// const util = require('util') // const util = require('util')
// console.log(util.inspect(status, false, null, true)) // console.log(util.inspect(status, false, null, true))
...@@ -187,35 +78,16 @@ module.exports = function (options = {}) { ...@@ -187,35 +78,16 @@ module.exports = function (options = {}) {
} }
else { else {
if (authstatus) { if (authstatus) {
require('./redis/on_register').on_register_redis(req.body.client_id).then(status => { on_register_redis(req.body.client_id)
console.log('redis:status: ' + status) res.send({ result: 'ok' });
if (status) {
res.send({
result: 'ok'
});
} else {
res.send({
result: {error: 'not allowed'}
});
}
})
} else { } else {
res.send({ res.send({ result: { error: 'not allowed' } });
result: {error: 'not allowed'}
>>>>>>> 9c233f79215090b0fd648f8b8b8061447ed746dc
});
} }
}) }
next(); next();
} }
else { else {
res.send({ res.send({ result: { error: 'not allowed' } });
<<<<<<< HEAD
result: { error: 'not allowed' }
=======
result: 'no'
>>>>>>> 9c233f79215090b0fd648f8b8b8061447ed746dc
});
next(); next();
} }
} }
......
authhook/on_deliver.js
...@@ -15,10 +15,10 @@ module.exports = function(options = {}) { ...@@ -15,10 +15,10 @@ module.exports = function(options = {}) {
console.log(response); console.log(response);
res.send(response) res.send(response)
} }
module.exports.on_deliver = on_deliver // module.exports.on_deliver = on_deliver
function joinTopic(topics) { // function joinTopic(topics) {
var topic = topics.split('/!')[1] // var topic = topics.split('/!')[1]
return topics.split('/!')[0] + topic.substring(topic.indexOf('/'), topic.length) // return topics.split('/!')[0] + topic.substring(topic.indexOf('/'), topic.length)
} // }
} }
authhook/on_offline.js
var on_offline_redis = require('./redis/on_offline').on_offline_redis
module.exports = function (options = {}) { module.exports = function (options = {}) {
debug = options.debug || false; debug = options.debug || false;
return function (req, res, next) { return function (req, res, next) {
console.log('auth_on_offline') console.log('auth_on_offline')
doRedis(req.body.client_id) on_offline_redis(req.body.client_id)
res.status(200); res.status(200);
res.send(''); res.send('');
} }
function doRedis(client_id) {
require('./redis/on_offline').on_offline_redis(client_id).then(status => {
console.log('redis:status: ' + status)
})
}
} }
\ No newline at end of file
authhook/redis/on_message.js
...@@ -2,27 +2,22 @@ var Redis = require('ioredis') ...@@ -2,27 +2,22 @@ var Redis = require('ioredis')
var redis = new Redis() var redis = new Redis()
function on_message_redis(deviceid, payload, topic) { // first time to access on authhook auth_on_register function on_message_redis(deviceid, payload, topic) { // first time to access on authhook auth_on_register
return new Promise((resolve, reject) => { if (deviceid.indexOf('mqtt') === 0) {
if (deviceid.indexOf('mqtt') === 0 || deviceid.length !== 36) { console.log(deviceid)
console.log(deviceid) }
resolve(true) else {
} var information = setValue(deviceid, payload, topic)
else { console.log(information.total_message_size)
var information = setValue(deviceid, payload, topic) redis.hincrby(information.keys, 'actual_message_count', 1)
console.log(information.total_message_size) redis.hincrby(information.keys, 'charged_message_count', information.charged_message_count)
redis.hincrby(information.keys, 'actual_message_count', 1) redis.hincrby(information.keys, 'total_message_size', Math.floor(information.total_message_size))
redis.hincrby(information.keys, 'charged_message_count', information.charged_message_count)
redis.hincrby(information.keys, 'total_message_size', Math.floor(information.total_message_size)) redis.hincrby(setBrokerValue().keys, 'actual_message_count', 1)
redis.hincrby(setBrokerValue().keys, 'charged_message_count', information.charged_message_count)
redis.hincrby(setBrokerValue().keys, 'total_message_size', Math.floor(information.total_message_size))
redis.hincrby(setBrokerValue().keys, 'actual_message_count', 1) redis.hset(information.keys, 'last_publish', information.last_publish)
redis.hincrby(setBrokerValue().keys, 'charged_message_count', information.charged_message_count) }
redis.hincrby(setBrokerValue().keys, 'total_message_size', Math.floor(information.total_message_size))
redis.hset(information.keys, 'last_publish', information.last_publish, function (err, res) {
resolve(!res) // 0 is ok, 1 is no => if ok is 0 then not 0 = 1
})
}
})
} }
module.exports.on_message_redis = on_message_redis module.exports.on_message_redis = on_message_redis
......
authhook/redis/on_offline.js
...@@ -2,22 +2,13 @@ var Redis = require('ioredis') ...@@ -2,22 +2,13 @@ var Redis = require('ioredis')
var redis = new Redis() var redis = new Redis()
function on_offline_redis(deviceid) { // first time to access on authhook auth_on_register function on_offline_redis(deviceid) { // first time to access on authhook auth_on_register
return new Promise((resolve, reject) => { if (deviceid.indexOf('mqtt') === 0) {
if (deviceid.indexOf('mqtt') === 0) { }
// if (deviceid.indexOf('mqtt') === 0 || deviceid.length !== 36) { else {
resolve(true) var information = setValue(deviceid)
} redis.hset(information.keys, 'status', information.status, 'offline_on', information.offline_on)
else { deviceUsageCount()
var information = setValue(deviceid) }
redis.hset(information.keys, 'status', information.status, 'offline_on', information.offline_on, function (err, res) {
console.log('res:', res)
if (res) {
deviceUsageCount()
resolve(true) // 0 is ok, 1 is no => if ok is 0 then not 0 = 1
} else resolve(false)
})
}
})
} }
module.exports.on_offline_redis = on_offline_redis module.exports.on_offline_redis = on_offline_redis
......
authhook/redis/on_register.js
...@@ -2,24 +2,15 @@ var Redis = require('ioredis') ...@@ -2,24 +2,15 @@ var Redis = require('ioredis')
var redis = new Redis() var redis = new Redis()
function on_register_redis(deviceid) { // first time to access on authhook auth_on_register function on_register_redis(deviceid) { // first time to access on authhook auth_on_register
return new Promise((resolve, reject) => { if (deviceid.indexOf('mqtt') === 0) {
// if (deviceid.indexOf('mqtt') === 0 || deviceid.length !== 36) { }
if (deviceid.indexOf('mqtt') === 0) { else {
resolve(true) var information = setValue(deviceid)
} redis.hdel(information.keys, 'offline_on')
else { console.log(information.keys)
var information = setValue(deviceid) redis.hset(information.keys, 'status', information.status, 'register_on', information.register_on, 'last_check', information.last_check)
redis.hdel(information.keys, 'offline_on') deviceUsageCount()
console.log(information.keys) }
redis.hset(information.keys, 'status', information.status, 'register_on', information.register_on, 'last_check', information.last_check, function (err, res) {
console.log('res: ',res)
// if (res) {
deviceUsageCount()
resolve(true) // 0 is ok, 1 is no => if ok is 0 then not 0 = 1
// } else resolve(false)
})
}
})
} }
module.exports.on_register_redis = on_register_redis module.exports.on_register_redis = on_register_redis
......
authhook/redis/on_status.js deleted 100644 → 0
var Redis = require('ioredis')
var redis = new Redis()
redis.subscribe('clientStatus')
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment