labs
/
playground-readmongo
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Browse Source

ok1

master
zeus 4 years ago
parent
93b3affe63
commit
fbb56567c5
1 changed files with 517 additions and 8 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 523
      swarmlab-app/src/run/app.js

523
swarmlab-app/src/run/app.js
View File

@ -18,12 +18,376 @@ const pubClient = RedisClient.createClient({
//const pubClient = new RedisClient({ host: 'localhost', port: 6379 });
const subClient = pubClient.duplicate();
//io.adapter(createAdapter({ pubClient, subClient }));
io.adapter(createAdapter({ pubClient, subClient }));
pubClient.on("connect", function() {
console.log("You are now connected");
});
const MongoClient = require('mongodb').MongoClient;
var async = require("async");
const { check, validationResult } = require('express-validator');
const urlExistSync = require("url-exist-sync");
var express = require('express');
app.use(express.json());
const axios = require('axios');
axios.defaults.timeout = 30000
const helmet = require('helmet');
app.use(helmet());
const cors = require('cors')
const whitelist = [
'http://localhost:8080',
'http://localhost:3080',
'http://localhost:3081',
'http://localhost:3082'
]
const corsOptions = {
credentials: true,
methods: ['GET', 'PUT', 'POST', 'DELETE', 'OPTIONS'],
optionsSuccessStatus: 200, // some legacy browsers (IE11, various SmartTVs) choke on 204
allowedHeaders: [
'Content-Type',
'Authorization',
'X-Requested-With',
'device-remember-token',
'Access-Control-Allow-Origin',
'Access-Control-Allow-Headers',
'Origin',
'Accept'
],
origin: function(origin, callback) {
if (whitelist.indexOf(origin) !== -1) {
callback(null, true)
} else {
callback(null, true)
//callback(new Error('Not allowed by CORS'))
}
}
}
// ***************************************************
// checktoken
// ***************************************************
async function checkToken(token) {
const agent = new https.Agent({
rejectUnauthorized: false,
});
const instance = axios.create({
baseURL: 'https://api.swarmlab.io',
withCredentials: true,
rejectUnauthorized: false,
crossdomain: true,
httpsAgent: agent,
headers: {
'Accept': 'application/json',
'Content-Type': 'multipart/form-data',
'Authorization': 'Bearer '+token
}
})
try {
var pipelines = {
"source":'ssologin'
}
var params = {
pipeline: pipelines
}
var options = {
headers: { 'content-type': 'application/x-www-form-urlencoded',Authorization: `Bearer ${token}` },
};
instance.defaults.timeout = 30000;
const res = await instance.post('/istokenvalidsso',params,options);
if(res.status == 200){
//console.log("check " +JSON.stringify(res.data))
return res.data
}else{
console.log("noerror: " + res)
return res.status
}
}
catch (err) {
console.error("error: "+err);
var error = new Object();
error.action = '401'
return error
}
}
function convertDateToUTC(date) {
return new Date(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate(), date.getUTCHours(), date.getUTCMinutes(), date.getUTCSeconds(),date.getUTCMilliseconds());
}
// ***************************************************
// get pipelines
// ***************************************************
async function getpipelines(token,pipelinename) {
const agent = new https.Agent({
rejectUnauthorized: false,
});
const instance = axios.create({
baseURL: 'https://api.swarmlab.io',
withCredentials: true,
rejectUnauthorized: false,
crossdomain: true,
httpsAgent: agent,
headers: {
'Accept': 'application/json',
'Content-Type': 'multipart/form-data',
'Authorization': 'Bearer '+token
}
})
/*
var params = {
playbook: value
}
var options = {
params: params,
headers: { 'content-type': 'application/x-www-form-urlencoded',Authorization: `Bearer ${token}` },
};
const playbook = await api.GET('playbookCode',options);
return playbook
*/
try {
var pipelines = {
"querytokenFilter":'uWr4FKRqrmpCRkJ9WLuI0DNuDWOGTkfcSzyZkJirZvJwwFDffLWrraqzzSPLeuQqL3TF9',
"filter":pipelinename
}
//var params = {
// pipeline: pipelines
// }
var params = {
querytokenFilter:'uWr4FKRqrmpCRkJ9WLuI0DNuDWOGTkfcSzyZkJirZvJwwFDffLWrraqzzSPLeuQqL3TF9',
filter:pipelinename
}
var options = {
params: params,
headers: { 'content-type': 'application/x-www-form-urlencoded',Authorization: `Bearer ${token}` },
};
//https://api.swarmlab.io/gettutorlabrooms?sort=pipelinename%7Casc&page=1&per_page=5&filter=&type=scripts&tutor=yes
instance.defaults.timeout = 30000;
//const res = await instance.get('/getplaygrounds',params,options);
const res = await instance.get('/getplaygrounds',options);
if(res.status == 200){
return res.data
}else{
console.log("noerror: " + res)
return res.status
}
}
catch (err) {
console.error("error: "+err);
var error = new Object();
error.action = '401'
return error
}
}
global.online='ob';
global.pipelines=[];
function sendlog(reslog,pathfileval){
var usertmp = global.pipelines.find(x => x.pathlogfile==pathfileval);
//for (var key in usertmp.data){
var user = usertmp.data[0].user25user;
// if(usertmp.data){
console.log('-----------------------' + JSON.stringify(usertmp));
io.in(user).emit("logdata", reslog);
// }
//}
}
function onlogfile(path){
console.log('File', path, 'has been added');
var pathfileval = pathmodule.basename(path);
var arrfile = pathfileval.toString().split("-");
var pathfile = arrfile[0];
var indexfind1 = global.pipelines.findIndex(x => x.pathlogfile==pathfileval);
console.log('file11111111111111111111111111111111 ' + JSON.stringify(pathfileval))
if (indexfind1 === -1 ){
(async() => {
console.log('file2222222222222222222222222222222222222 ' + JSON.stringify(pathfileval))
var token = "d2539e5a7ae1f9f1b0eb2b8f22ca467a86d28407"; // desto
var resdata = await getpipelines(token,pathfile)
//resdata.data.pathlogfile = 'test'
var resob = {}
resob.pathlogfile = pathfileval
var resobarray = []
for (let i in resdata.data) {
var resob1 = {}
resob1.data = resdata.data[i].res25swarmlabname
resob1.user25user = resdata.data[i].res25user
resob1.res25creator = resdata.data[i].res25creator
resob1.res25fileforce = resdata.data[i].res25fileforce
resobarray.push(resob1)
}
resob.data = resobarray
var indexfind = global.pipelines.findIndex(x => x.pathlogfile==pathfileval);
indexfind === -1 ? global.pipelines.push(resob) : console.log("object already exists "+pathfileval)
})()
}
}
// ***************************************************
// rest get
// ***************************************************
app.get('/run', [
//check('access_token').isLength({ min: 40 }),
//check('llo').isBase64()
],
cors(corsOptions), (req, res, next) => {
(async() => {
var mongourl = "mongodb://playgrounduser:efvvnuioervefSDFSGYGHRDFVsdfergvssppiiedifhwincvinviw_dbfjbsifbsdkjfswuunscfudfgbbfvibqefwrvnine@ondemand_playground_mongo1:27017,ondemand_playground_mongo2:27017,ondemand_playground_mongo3:27017,ondemand_playground_mongo4:27017,ondemand_playground_mongo5:27017,ondemand_playground_mongo6:27017,ondemand_playground_mongo7:27017/fluent?replicaSet=rs1"
const OPTS = {
useNewUrlParser: true,
useUnifiedTopology: true
};
MongoClient.connect(mongourl, OPTS, function(err, client){
if(err){
console.log(err);
} else {
const db = client.db('fluent');
//db.collection('log', onCollection);
console.log(JSON.stringify('mongo connected'))
var stream = db.collection('logs').find({}, {
tailable: true,
awaitdata: true
/* other options */
}).stream();
stream.on('data', function (doc) {
console.log(JSON.stringify(doc))
//socket.write(JSON.stringify({'action': 'log','param': doc.log}));
});
}
});
var RES = new Object();
RES.code = req.query["filter"]
RES.token = req.query["filter"]
var isvalid = await checkToken(RES.token);
if(isvalid.action == 'ok'){
console.log("Authserver ok " + RES.token);
}else{
console.log("Authserver no " + RES.token);
}
RES.error = false
RES.error_msg = "ok"
res.json(RES)
})()
});
// ***************************************************
// rest post
// ***************************************************
app.post('/run', [
//check('access_token').isLength({ min: 40 }),
//check('llo').isBase64()
],
cors(corsOptions), (req, res, next) => {
(async() => {
//console.log(JSON.stringify(req.headers));
//console.log(JSON.stringify(req.body));
//console.log("mongo "+JSON.stringify(req.body));
//console.log("LOG "+JSON.stringify(req.body[0].message));
//console.log("PATH "+JSON.stringify(req.body[0].tailed_path));
for (var i = 0; i < req.body.length; i++){
//var getpath = await onlogfile(req.body[i].tailed_path)
var path = req.body[i].tailed_path
console.log('File', path, 'has been added');
var pathfileval = pathmodule.basename(path);
var arrfile = pathfileval.toString().split("-");
var pathfile = arrfile[0];
var indexfind1 = global.pipelines.findIndex(x => x.pathlogfile==pathfileval);
console.log('file11111111111111111111111111111111 ' + JSON.stringify(pathfileval))
if (indexfind1 === -1 ){
(async() => {
console.log('file2222222222222222222222222222222222222 ' + JSON.stringify(pathfileval))
var token = "d2539e5a7ae1f9f1b0eb2b8f22ca467a86d28407"; // desto
var resdata = await getpipelines(token,pathfile)
//resdata.data.pathlogfile = 'test'
var resob = {}
resob.pathlogfile = pathfileval
var resobarray = []
for (let i in resdata.data) {
var resob1 = {}
resob1.data = resdata.data[i].res25swarmlabname
resob1.user25user = resdata.data[i].res25user
resob1.res25creator = resdata.data[i].res25creator
resob1.res25fileforce = resdata.data[i].res25fileforce
resobarray.push(resob1)
}
resob.data = resobarray
//var indexfind = global.pipelines.findIndex(x => x.res25swarmlabname==resdata.data[0].res25swarmlabname);
var indexfind = global.pipelines.findIndex(x => x.pathlogfile==pathfileval);
//indexfind === -1 ? global.pipelines.push({resob}) : console.log("object already exists")
indexfind === -1 ? global.pipelines.push(resob) : console.log("object already exists "+pathfileval)
//console.log('info', JSON.stringify(resdata));
//console.log('info------------- ', JSON.stringify(global.pipelines));
})()
}
//
var obj = req.body[i];
//var indexfind = global.pipelines.findIndex(x => x.res25swarmlabname==resdata.data.res25swarmlabname);
//indexfind === -1 ? global.pipelines.push(resob) : console.log("object already exists")
var now = new Date();
var reslog = new Object();
reslog.log = obj
reslog.date = convertDateToUTC(now)
console.log(reslog);
var pathfileval = pathmodule.basename(reslog.log.tailed_path);
var indexfind = global.pipelines.findIndex(x => x.pathlogfile==pathfileval);
//indexfind === -1 ? sendlog(reslog,pathfileval) : console.log("object already exists")
indexfind === -1 ? console.log("object not found") : sendlog(reslog,pathfileval)
console.log("IOT "+JSON.stringify(reslog.log.tailed_path));
console.log("IOTindexfind "+JSON.stringify(indexfind));
console.log("IOTuser "+JSON.stringify(global.pipelines));
// io.in("iot").emit("message", reslog);
// io.emit("logdata", reslog);
}
})()
//io.in("iot").emit("message", RES);
console.error('socket POST from client');
var RES = new Object();
RES.error = false
RES.error_msg = "ok"
RES.msg = req.body[0].messsage
res.json(RES)
});
// ***************************************************
// socket
@ -31,18 +395,163 @@ pubClient.on("connect", function() {
io.origins('*:*') // for latest version
var itemsProcessed = 0;
function onCollection(err, collection) {
let options = { tailable: true,
awaitdata: true,
numberOfRetries: -1,
tailableRetryInterval: 500
};
var cursor = collection.find({},options).stream();
var itemsProcessed = 0;
var reslog = new Object();
var now = new Date();
cursor.on('data', function (data) {
console.log('+++++++++ondata<<<<<<<<<<<<<<<<<<<<<<' + data);
io.in('anagnostopoulos@uniwa.gr').emit("logdata", data);
var pathfileval = pathmodule.basename(data.tailed_path);
var arrfile = pathfileval.toString().split("-");
var pathfile = arrfile[0];
var indexupdate = true
var resob = {}
//var indexfind1 = global.pipelines.findIndex(x => x.pathlogfile==pathfileval);
pubClient.get(pathfileval, function(err, object) {
if(object){
indexupdate = false
}else{
console.log('redis '+JSON.stringify(object));
}
});
if (indexupdate ){
(async() => {
var token = "d2539e5a7ae1f9f1b0eb2b8f22ca467a86d28407"; // desto
var resdata = await getpipelines(token,pathfile)
resob.pathlogfile = pathfileval
var resobarray = []
var resob1 = {}
var i = 0
resob1.data = resdata.data[i].res25swarmlabname
resob1.user25user = resdata.data[i].res25user
resob1.res25creator = resdata.data[i].res25creator
resob1.res25fileforce = resdata.data[i].res25fileforce
resob1.tailed_path = pathfileval
//resobarray.push(resob1)
var resob1string = JSON.stringify(resob1);
console.log('+++++++++++++++++' + resob1string + '<<<<<<<<<<<<<<<<<<<<<<' + pathfileval);
pubClient.set(pathfileval, resob1string, function(err, res) {
});
reslog.data = resob1
reslog.log = data
reslog.date = convertDateToUTC(now)
var user = resob1.res25creator
console.log('datauser ' + JSON.stringify(user));
//io.in(user).emit("logdata", reslog);
var user = 'anagnostopoulos@uniwa.gr'
io.in(user).emit("logdata", reslog);
itemsProcessed++;
})()
}else{
pubClient.get(pathfileval, function(err, object) {
var resob1 = {}
var i = 0
resob1.data = object.res25swarmlabname
resob1.user25user = object.res25user
resob1.res25creator = object.res25creator
resob1.res25fileforce = object.res25fileforce
resob1.tailed_path = object.tailed_path
reslog.data = resob1
reslog.log = data
reslog.date = convertDateToUTC(now)
console.log('>>>>>>>>>>2222222222222>>>>>>>>>>>>>>>>>>>>>>>>>>>>> '+JSON.stringify(object));
var user = object.user25user
io.in(user).emit("logdata", reslog);
itemsProcessed++;
});
}
});
setInterval(function () {
var user = 'anagnostopoulos@uniwa.gr'
io.in(user).emit("logdata", 'test');
console.log('itemsProcessed', itemsProcessed++);
console.log('itemsProcessed', itemsProcessed);
}, 1000);
}
var mongourl = "mongodb://playgrounduser:efvvnuioervefSDFSGYGHRDFVsdfergvssppiiedifhwincvinviw_dbfjbsifbsdkjfswuunscfudfgbbfvibqefwrvnine@ondemand_playground_mongo1:27017,ondemand_playground_mongo2:27017,ondemand_playground_mongo3:27017,ondemand_playground_mongo4:27017,ondemand_playground_mongo5:27017,ondemand_playground_mongo6:27017,ondemand_playground_mongo7:27017/fluent?replicaSet=rs1"
const OPTS = {
useNewUrlParser: true,
useUnifiedTopology: true
};
var mongooptions = {
autoReconnect: true,
keepAlive: 1,
connectTimeoutMS: 30000,
socketTimeoutMS: 0
}
MongoClient.connect(mongourl, OPTS, function(err, client){
if(err){
console.log(err);
} else {
const db = client.db('fluent');
db.collection('logs', onCollection);
}
});
io.on('connection', s => {
console.error('socket connection');
var room = 'anagnostopoulos@uniwa.gr'
s.join(room);
// ------------------------------
// --- set
// ------------------------------
var usersession = new Object();
usersession.SOCKET = {};
usersession.SOCKET.error = {};
console.error('socket ...');
s.auth = false;
// ------------------------------
// --- authenticate
// ------------------------------
s.on('authenticate', function(data){
const token = data
console.log('invalid 1 ' + token);
(async() => {
var isvalid = await checkToken(token);
if(isvalid.action == 'ok'){
console.log("Authserver ok ", s.id + ' - ' + token);
usersession.SOCKET.user = isvalid.user
usersession.SOCKET.scope = isvalid.scope // space delimeter
usersession.SOCKET.token = isvalid.token
s.auth = true;
}else{
console.log("Authserver no ", s.id + ' - ' + token);
s.auth = false;
}
})()
});
setTimeout(function(){
if (!s.auth) {
console.log("Disconnecting timeout socket ", s.id);
s.disconnect('unauthorized');
}else{
var room = usersession.SOCKET.user
//s.on("subscribe", function (room) {
// console.log("joining room", room);
s.join(room);
// });
}
}, 30000);
var id = s.id
s.on('log', obj => {
console.error('from client '+ s.id + ' obj ' + obj);

Write Preview
Loading…
Cancel
Save