Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Integrate refactored models #8

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Integrate refactored models #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5af0be3
Integrate refactored models
Aug 21, 2017
a8c6c69
Channel constructor receives an id as paramater
Aug 23, 2017
7e26c3b
Cleaned code
Aug 23, 2017
4443972
Use functions from models
Aug 25, 2017
edff74e
Use messaging client initalized in models
Aug 25, 2017
635a99f
Use function from models
Aug 25, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 28 additions & 100 deletions index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ var async = require('async');
var redis = require('redis');
require('colors');

var Models = require('telepat-models');
var tlib = require('telepat-models');

var workerType = args.params.t;
var workerIndex = args.params.i;
Expand All @@ -12,134 +12,62 @@ var workerIndex = args.params.i;
* @type {Base_Worker}
*/
var theWorker = null;
var configManager = null;

async.series([
function(callback) {
configManager = new Models.ConfigurationManager('./config.spec.json', './config.json');
configManager.load(function(err) {
if (err) return callback(err);

var testResult = configManager.test();
callback(testResult !== true ? testResult : undefined);
});
seriesCallback => {
let extraObj = {
name: workerType + '-' + workerIndex,
broadcast: workerType === 'sockets_transport'? true:false,
exclusive: false,
type: workerType
};
tlib.init(extraObj, seriesCallback, workerType);
},
function(callback) {
(seriesCallback) => {
switch (workerType) {
case 'aggregation': {
case 'aggregation': {
var AggregationWorker = require('./lib/aggregation_worker');
theWorker = new AggregationWorker(workerIndex, configManager.config);

theWorker = new AggregationWorker(workerIndex, tlib.config);

break;
}
case 'write': {
var WriterWorker = require('./lib/writer_worker');
theWorker = new WriterWorker(workerIndex, configManager.config);

theWorker = new WriterWorker(workerIndex, tlib.config);
theWorker.redisClient = tlib.services.redisClient;
theWorker.redisCacheClient = tlib.services.redisCacheClient;

break;
}
case 'transport_manager': {
var TransportManagerWorker = require('./lib/transport_manager');
theWorker = new TransportManagerWorker(workerIndex, configManager.config);

theWorker = new TransportManagerWorker(workerIndex, tlib.config);

break;
}
default: {
var workerTypeParts = workerType.split('_');

if (workerTypeParts[1] === 'transport') {
var ClientTransportWorker = require('./lib/client_transport/'+workerTypeParts[0]);
theWorker = new ClientTransportWorker(workerIndex, configManager.config);
var ClientTransportWorker = require('./lib/client_transport/' + workerTypeParts[0]);

theWorker = new ClientTransportWorker(workerIndex, tlib.config);
} else {
console.log('Invalid worker type "'+workerType+'"');
console.log('Invalid worker type "' + workerType + '"');
process.exit(1);
}
}
}
callback();
},
function(callback) {
theWorker.config.subscribe_limit = theWorker.config.subscribe_limit || 64;
theWorker.config.get_limit = theWorker.config.get_limit || 384;

if (theWorker.config.logger) {
theWorker.config.logger.name = theWorker.name;
Models.Application.logger = new Models.TelepatLogger(theWorker.config.logger);
} else {
Models.Application.logger = new Models.TelepatLogger({
type: 'Console',
name: theWorker.name,
settings: {level: 'info'}
});
}

if (!Models[theWorker.config.main_database]) {
Models.Application.logger.emergency('Unable to load "'+theWorker.config.main_database+
'" main database: not found. Aborting...');
process.exit(2);
}

Models.Application.datasource = new Models.Datasource();
Models.Application.datasource.setMainDatabase(new Models[theWorker.config.main_database](theWorker.config[theWorker.config.main_database]));

callback();
},
function(callback) {
Models.Application.datasource.dataStorage.onReady(function() {
callback();
});
},
function(callback) {
if (Models.Application.redisClient)
Models.Application.redisClient = null;

Models.Application.redisClient = redis.createClient(theWorker.config.redis.port, theWorker.config.redis.host);
Models.Application.redisClient.on('error', function(err) {
Models.Application.logger.error('Failed connecting to Redis "'+
theWorker.config.redis.host+'": '+err.message+'. Retrying...');
});
Models.Application.redisClient.on('ready', function() {
Models.Application.logger.info('Client connected to Redis.');
callback();
});
},
function(callback) {
if (Models.Application.redisCacheClient)
Models.Application.redisCacheClient = null;
seriesCallback();

Models.Application.redisCacheClient = redis.createClient(theWorker.config.redisCache.port, theWorker.config.redisCache.host);
Models.Application.redisCacheClient.on('error', function(err) {
Models.Application.logger.error('Failed connecting to Redis Cache "'+theWorker.config.redisCache.host+'": '+
err.message+'. Retrying...');
});
Models.Application.redisCacheClient.on('ready', function() {
Models.Application.logger.info('Client connected to Redis Cache.');
callback();
});
},
function(callback) {
if (!Models[theWorker.config.message_queue]) {
Models.Application.logger.emergency('Unable to load "'+theWorker.config.message_queue+
'" messaging queue: not found. Aborting...');
process.exit(-1);
}

var messageQueueConfig = theWorker.config[theWorker.config.message_queue];

if (messageQueueConfig === undefined) {
messageQueueConfig = {broadcast: theWorker.broadcast, exclusive: theWorker.exclusive};
} else {
messageQueueConfig.broadcast = theWorker.broadcast;
messageQueueConfig.exclusive = theWorker.exclusive;
}

var messagingClient = new Models[theWorker.config.message_queue](messageQueueConfig, theWorker.name, workerType);
theWorker.messagingClient = messagingClient;

messagingClient.onReady(callback);
}
], function(err) {
], function (err) {
if (err) {
throw err;
}

theWorker.ready();
});
});
Loading