Skip to content

Commit 5a80829

Browse files
Gennaro Del SorboGennaro Del Sorbo
Gennaro Del Sorbo
authored and
Gennaro Del Sorbo
committed
refactor
1 parent cd406c6 commit 5a80829

File tree

3 files changed

+152
-113
lines changed

3 files changed

+152
-113
lines changed

‎Application/Utils.js

+13
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
"use strict"
22

3+
const _ = require("underscore")
4+
35
class Utils {
46
constructor() {}
57

@@ -8,6 +10,17 @@ class Utils {
810
return obj;
911
return other;
1012
}
13+
14+
isNull(val) {
15+
return val === null || val === undefined
16+
}
17+
18+
areNotNull() {
19+
20+
return _.find(_.values(arguments), val => {
21+
return this.isNull(val)
22+
}) === undefined
23+
}
1124
}
1225

1326
module.exports = new Utils();

‎Partitioner.js

+89-94
Original file line numberDiff line numberDiff line change
@@ -1,119 +1,114 @@
1-
var cluster = require('cluster');
2-
var jobService = require("./Services/JobService");
3-
var PartitionService = require("./Services/PartitionService");
4-
var jobs = require("./Application/Jobs");
5-
var Lock = require("./Application/ExecuteLocked");
6-
var lock = new Lock();
7-
var Worker = require("./Application/Worker");
8-
var validator = require("validator");
9-
var utils = require("./Application/Utils");
10-
var variables = require("./Application/CommonVariables");
1+
"use strict"
112

12-
var workers = [];
13-
var workerPartitionIndex = 0;
14-
var numberOfWorkers;
15-
var logger;
3+
const cluster = require('cluster');
4+
const jobService = require("./Services/JobService");
5+
const PartitionService = require("./Services/PartitionService");
6+
const jobs = require("./Application/Jobs");
7+
const Lock = require("./Application/ExecuteLocked");
8+
const lock = new Lock();
9+
const Worker = require("./Application/Worker");
10+
const validator = require("validator");
11+
const utils = require("./Application/Utils");
12+
const variables = require("./Application/CommonVariables");
1613

17-
var defaultConfiguration = {
18-
numberOfWorkers: 1,
19-
cleanIdlePartitionsAfterMinutes: 15,
20-
loggerLevel: "error",
21-
consoleLogger: true,
22-
fileLogger: true,
23-
fileLoggerPath: "./logger"
14+
let _workers = [];
15+
let _workerPartitionIndex = 0;
16+
let _numberOfWorkers;
17+
let _logger;
18+
19+
const defaultConfiguration = {
20+
numberOfWorkers: 1,
21+
cleanIdlePartitionsAfterMinutes: 15,
22+
loggerLevel: "error",
23+
consoleLogger: true,
24+
fileLogger: true,
25+
fileLoggerPath: "./logger"
2426
};
2527

26-
function Partitioner(configuration) {
27-
if(cluster.isWorker)
28-
throw new Error("a worker is trying to instantiate a partitioner");
29-
30-
if(configuration !== undefined)
31-
validate(configuration);
32-
33-
var config = configuration !== undefined ? configuration : defaultConfiguration;
34-
numberOfWorkers = utils.coalesce(config.numberOfWorkers, defaultConfiguration.numberOfWorkers);
35-
this.partitionService = new PartitionService(utils.coalesce(config.cleanIdlePartitionsAfterMinutes, defaultConfiguration.cleanIdlePartitionsAfterMinutes));
36-
37-
var processEnv = {};
38-
39-
var Logger = require("./Application/Logger");
40-
var consoleLogger = utils.coalesce(config.consoleLogger, defaultConfiguration.consoleLogger);
41-
var fileLogger = utils.coalesce(config.fileLogger, defaultConfiguration.fileLogger);
42-
var fileLoggerPath = utils.coalesce(config.fileLoggerPath, defaultConfiguration.fileLoggerPath);
43-
var loggerLevel = utils.coalesce(config.loggerLevel, defaultConfiguration.loggerLevel);
44-
Logger.new(consoleLogger, loggerLevel, fileLogger, fileLoggerPath).then(function(log) {
45-
logger = log;
46-
processEnv[variables.loggerLevel] = loggerLevel;
47-
processEnv[variables.consoleLogger] = consoleLogger;
48-
processEnv[variables.fileLogger] = fileLogger;
49-
processEnv[variables.fileLoggerPath] = fileLoggerPath;
50-
51-
for(var i=0; i < numberOfWorkers; i++){
52-
workers.push(new Worker(cluster.fork(processEnv)));
28+
class Partitioner {
29+
constructor(configuration) {
30+
if (cluster.isWorker)
31+
throw new Error("a worker is trying to instantiate a partitioner")
32+
33+
if (configuration)
34+
validate(configuration)
35+
36+
const config = configuration ? configuration : defaultConfiguration
37+
_numberOfWorkers = utils.coalesce(config.numberOfWorkers, defaultConfiguration.numberOfWorkers)
38+
39+
this.partitionService = new PartitionService(utils.coalesce(config.cleanIdlePartitionsAfterMinutes, defaultConfiguration.cleanIdlePartitionsAfterMinutes))
40+
41+
let processEnv = {}
42+
const Logger = require("./Application/Logger")
43+
const consoleLogger = utils.coalesce(config.consoleLogger, defaultConfiguration.consoleLogger)
44+
const fileLogger = utils.coalesce(config.fileLogger, defaultConfiguration.fileLogger)
45+
const fileLoggerPath = utils.coalesce(config.fileLoggerPath, defaultConfiguration.fileLoggerPath)
46+
const loggerLevel = utils.coalesce(config.loggerLevel, defaultConfiguration.loggerLevel)
47+
Logger.new(consoleLogger, loggerLevel, fileLogger, fileLoggerPath).then(log => {
48+
_logger = log
49+
processEnv[variables.loggerLevel] = loggerLevel
50+
processEnv[variables.consoleLogger] = consoleLogger
51+
processEnv[variables.fileLogger] = fileLogger
52+
processEnv[variables.fileLoggerPath] = fileLoggerPath
53+
54+
for (var i = 0; i < _numberOfWorkers; i++) {
55+
_workers.push(new Worker(cluster.fork(processEnv)));
5356
}
54-
}
55-
);
56-
}
57+
})
58+
}
5759

58-
Partitioner.prototype.enqueueJob = function(job, callback){
59-
var self = this;
60-
if(job === null
61-
|| job === undefined
62-
|| job.id === null
63-
|| job.id === undefined
64-
|| job.partitionId === null
65-
|| job.partitionId === undefined
66-
|| job.type === null
67-
|| job.type === undefined)
68-
throw new Error("Job null or invalid, should contain id, partitionId, type, data: {}");
69-
70-
lock.execWrite(function(){
71-
return self.partitionService.get(job.partitionId)
72-
.then(function(partition){
73-
if(partition == null) {
74-
var index = ++workerPartitionIndex % numberOfWorkers;
75-
return self.partitionService.push(job.partitionId, workers[index].worker);
76-
}else{
77-
return partition;
78-
}
79-
});
80-
}).then(function(partition){
81-
jobService.push(job.id, callback).then(function(){
82-
logger.debug("jobId: %d, partitionId: %d, type: %s, pushed", job.id, job.partitionId, job.type);
83-
partition.worker.send(job);
84-
});
85-
});
86-
};
60+
enqueueJob(job, callback) {
61+
if (!utils.areNotNull(job, job.id, job.partitionId, job.type))
62+
throw new Error("Job null or invalid, should contain id, partitionId, type, data: {}")
63+
64+
lock.execWrite(() => {
65+
return this.partitionService.get(job.partitionId)
66+
.then(partition => {
67+
if (utils.isNull(partition)) {
68+
const index = ++_workerPartitionIndex % _numberOfWorkers
69+
return this.partitionService.push(job.partitionId, _workers[index].worker)
70+
} else {
71+
return partition
72+
}
73+
})
74+
}).then(partition => {
75+
jobService.push(job.id, callback).then(() => {
76+
_logger.debug("jobId: %d, partitionId: %d, type: %s, pushed", job.id, job.partitionId, job.type)
77+
partition.worker.send(job)
78+
})
79+
})
80+
}
81+
}
8782

88-
function validate(configuration){
89-
if(configuration.numberOfWorkers !== undefined && !validator.isInt(configuration.numberOfWorkers, {min:1} ))
83+
function validate(configuration) {
84+
if (configuration.numberOfWorkers !== undefined && !validator.isInt(configuration.numberOfWorkers, { min: 1 }))
9085
throw new Error("numberOfWorkers should be an integer >= 1");
91-
if(configuration.cleanIdlePartitionsAfterMinutes !== undefined && !validator.isInt(configuration.cleanIdlePartitionsAfterMinutes, {min:1} ))
86+
if (configuration.cleanIdlePartitionsAfterMinutes !== undefined && !validator.isInt(configuration.cleanIdlePartitionsAfterMinutes, { min: 1 }))
9287
throw new Error("cleanIdlePartitionsAfterMinutes should be an integer >= 1");
93-
if(configuration.loggerLevel !== undefined && !(
94-
validator.equals(configuration.loggerLevel, 'debug')
88+
if (configuration.loggerLevel !== undefined && !(
89+
validator.equals(configuration.loggerLevel, 'debug')
9590
|| validator.equals(configuration.loggerLevel, 'info')
9691
|| validator.equals(configuration.loggerLevel, 'warn')
9792
|| validator.equals(configuration.loggerLevel, 'error'))
98-
)
93+
)
9994
throw new Error("loggerLevel should be debug, info, warn or error");
100-
if(configuration.consoleLogger !== undefined && !(
101-
validator.equals(configuration.consoleLogger, true)
95+
if (configuration.consoleLogger !== undefined && !(
96+
validator.equals(configuration.consoleLogger, true)
10297
|| validator.equals(configuration.consoleLogger, false))
103-
)
98+
)
10499
throw new Error("consoleLogger should be true or false");
105-
if(configuration.fileLogger !== undefined && !(
106-
validator.equals(configuration.fileLogger, true)
100+
if (configuration.fileLogger !== undefined && !(
101+
validator.equals(configuration.fileLogger, true)
107102
|| validator.equals(configuration.fileLogger, false))
108-
)
103+
)
109104
throw new Error("fileLogger should be true or false");
110-
if(configuration.fileLoggerPath !== undefined && typeof(configuration.fileLoggerPath) !== typeof(defaultConfiguration.fileLoggerPath))
105+
if (configuration.fileLoggerPath !== undefined && typeof (configuration.fileLoggerPath) !== typeof (defaultConfiguration.fileLoggerPath))
111106
throw new Error("fileLoggerPath should be a string");
112107
}
113108

114109
module.exports = {
115110
Partitioner: Partitioner,
116-
registerJob: function(title, func) {
111+
registerJob: function (title, func) {
117112
jobs[title] = func;
118113
}
119114
};

‎Test/Application/UtilsTest.js

+50-19
Original file line numberDiff line numberDiff line change
@@ -4,27 +4,58 @@ require("should");
44
const utils = require("../../Application/Utils")
55

66
describe("Utils Test", () => {
7-
it("If first is null, get the second", () => {
8-
let first = null;
9-
let second = 1;
10-
utils.coalesce(first, second).should.equal(second);
11-
})
7+
8+
describe("Coalesce", () => {
9+
it("If first is null, get the second", () => {
10+
let first = null;
11+
let second = 1;
12+
utils.coalesce(first, second).should.equal(second);
13+
})
1214

13-
it("If first is undefined, get the second", () => {
14-
let first;
15-
let second = 1;
16-
utils.coalesce(first, second).should.equal(second);
17-
})
15+
it("If first is undefined, get the second", () => {
16+
let first;
17+
let second = 1;
18+
utils.coalesce(first, second).should.equal(second);
19+
})
1820

19-
it("If first is false, get first", () => {
20-
let first = false;
21-
let second = 1;
22-
utils.coalesce(first, second).should.equal(first);
23-
})
21+
it("If first is false, get first", () => {
22+
let first = false;
23+
let second = 1;
24+
utils.coalesce(first, second).should.equal(first);
25+
})
2426

25-
it("If first is not null, get first", () => {
26-
let first = 1;
27-
let second = 2;
28-
utils.coalesce(first, second).should.equal(first);
27+
it("If first is not null, get first", () => {
28+
let first = 1;
29+
let second = 2;
30+
utils.coalesce(first, second).should.equal(first);
31+
})
32+
})
33+
34+
describe("IsNull", () => {
35+
it("If is null, return true", () => {
36+
utils.isNull(null).should.be.true()
37+
})
38+
39+
it("If is undefined, return true", () => {
40+
utils.isNull(undefined).should.be.true()
41+
})
42+
43+
it("If is not null or undefined, return false", () => {
44+
utils.isNull(false).should.be.false()
45+
})
46+
})
47+
48+
describe("AreNotNull", () => {
49+
it("If one is null, return false", () => {
50+
utils.areNotNull(1, null).should.be.false()
51+
})
52+
53+
it("If one is undefined, return false", () => {
54+
utils.areNotNull(undefined, 1).should.be.true()
55+
})
56+
57+
it("If is not null or undefined, return true", () => {
58+
utils.areNotNull(1, 2, 3).should.be.true()
59+
})
2960
})
3061
});

0 commit comments

Comments
 (0)