Skip to content

Commit d3470ae

Browse files
JexSrsJexSrs
committedJul 4, 2022
Finished grid
Tests are required
·
v6.2.1v3.0.0
1 parent 5703f91 commit d3470ae

File tree

4 files changed

+94
-37
lines changed

4 files changed

+94
-37
lines changed
 

‎src/components/config.ts

+26-4
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,21 @@ export type ConfigType = {
2828
checksInterval?: number;
2929
};
3030
grid: {
31-
distributed: boolean;
32-
serverAddress?: string;
31+
distributed: false;
32+
} | {
33+
distributed: true;
34+
serverAddress: string;
35+
serverPort?: number;
36+
authToken: string;
37+
useHTTP: false;
38+
} | {
39+
distributed: true;
40+
serverAddress: string;
3341
serverPort?: number;
42+
authToken: string;
43+
useHTTP: true;
44+
key: any;
45+
cert: any;
3446
};
3547
misc: {
3648
log?: 'all' | 'info' | 'errors' | 'none';
@@ -66,8 +78,10 @@ export default class Config {
6678
},
6779
grid: {
6880
distributed: false,
69-
serverAddress: undefined,
81+
// For default values on false
82+
// @ts-ignore
7083
serverPort: 3000,
84+
useHTTP: false
7185
},
7286
misc: {
7387
log: "all"
@@ -146,8 +160,16 @@ export default class Config {
146160
return !isStatic ? Config.load().grid.distributed : false;
147161
case ConfigOptions.GRID_SERVER_ADDRESS:
148162
return !isStatic ? Config.load().grid.serverAddress : 'localhost';
149-
case ConfigOptions.GRID_SERVER_PORT:
163+
case ConfigOptions.GRID_SERVER_PORT:
150164
return !isStatic ? Config.load().grid.serverPort : 3000;
165+
case ConfigOptions.GRID_AUTH:
166+
return !isStatic ? Config.load().grid.authToken : undefined;
167+
case ConfigOptions.GRID_USE_HTTP:
168+
return !isStatic ? Config.load().grid.useHTTP : false;
169+
case ConfigOptions.GRID_HTTPS_KEY:
170+
return !isStatic ? Config.load().grid.key : undefined;
171+
case ConfigOptions.GRID_HTTPS_CERT:
172+
return !isStatic ? Config.load().grid.cert : undefined;
151173

152174
case ConfigOptions.MISC_LOG_LEVEL:
153175
return !isStatic ? Config.load().misc.log : "all";

‎src/middleware/ConfigOptions.ts

+4
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,10 @@ export enum ConfigOptions {
1212
GRID_DISTRIBUTED = 'grid.distributed',
1313
GRID_SERVER_ADDRESS = 'grid.server.address',
1414
GRID_SERVER_PORT = 'grid.server.port',
15+
GRID_AUTH = 'grid.auth',
16+
GRID_USE_HTTP = 'grid.use_http',
17+
GRID_HTTPS_KEY = 'grid.https.key',
18+
GRID_HTTPS_CERT = 'grid.https.cert',
1519
MISC_LOG_LEVEL = 'misc.log',
1620
DB_PUSH_ARTICLES = 'db.articles.push',
1721
DB_GET_ARTICLES = 'db.articles.get'

‎src/modules/events.ts

+4-4
Original file line numberDiff line numberDiff line change
@@ -65,10 +65,10 @@ export default class Events {
6565
this.getAntennae().on("grid.worker.destroyed", (worker_id: string) =>
6666
Logger(LoggerTypes.DEBUG, `${chalk.yellow('Grid')} - worker destroyed (${worker_id}).`));
6767

68-
this.getAntennae().on("workers.job.finished", (job: Job) =>
69-
Logger(LoggerTypes.DEBUG, `${chalk.green('Worker')} - finished job(${job.id}).`));
70-
this.getAntennae().on("workers.job.failed", (job: Job) =>
71-
Logger(LoggerTypes.DEBUG, `${chalk.green('Worker')} - failed job(${job.id}).`));
68+
this.getAntennae().on("workers.job.finished", (jobId: string) =>
69+
Logger(LoggerTypes.DEBUG, `${chalk.green('Worker')} - finished job(${jobId}).`));
70+
this.getAntennae().on("workers.job.failed", (jobId: string) =>
71+
Logger(LoggerTypes.DEBUG, `${chalk.green('Worker')} - failed job(${jobId}).`));
7272

7373
this.getAntennae().on("workers.articles.found", (articles: Article[], src: string) =>
7474
Logger(LoggerTypes.DEBUG, `${chalk.cyan('Articles')} - Finished job returned ${articles.length} articles for ${src}.`));

‎src/modules/grid/index.ts

+60-29
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,15 @@ import Events from "../events";
33
import {JobStatus} from "../../components/JobStatus";
44
import Worker from "../workers";
55
import Config from "../../components/config";
6-
import {nanoid} from "nanoid";
76
import Article from "../../components/articles";
87
import {ConfigOptions} from "../../middleware/ConfigOptions";
98
import Extensions from "../extensions";
109
import Source from "../../components/source";
1110
import Scheduler from "../scheduler";
1211
import * as ServerIO from "socket.io"
1312
import * as ClientIO from "socket.io-client";
13+
import * as http from "http";
14+
import * as https from "https";
1415

1516

1617
export default class Grid {
@@ -19,48 +20,71 @@ export default class Grid {
1920
private declare readonly isMain: boolean
2021
private declare readonly workersIds: string[];
2122
private declare workersClients: { workersIds: string[], socketId: string }[];
22-
private declare readonly encryptionKey: string
2323

24-
private declare server: ServerIO.Server;
25-
private declare client: ClientIO.Socket;
24+
private declare readonly http_s_server: any;
25+
private declare readonly server: ServerIO.Server;
26+
private declare readonly client: ClientIO.Socket;
2627

2728
private constructor() {
28-
this.isMain = Config.getOption(ConfigOptions.SAFFRON_MODE) === 'main'
29-
this.workersIds = []
30-
this.workersClients = []
31-
this.encryptionKey = nanoid(256)
29+
this.isMain = Config.getOption(ConfigOptions.SAFFRON_MODE) === 'main';
30+
this.workersIds = [];
31+
this.workersClients = [];
3232

33-
if (this.isMain)
34-
this.server = new ServerIO.Server({
35-
// TODO
33+
if(Config.getOption(ConfigOptions.GRID_DISTRIBUTED)) {
34+
if (typeof Config.getOption(ConfigOptions.GRID_AUTH) !== 'string')
35+
throw new Error('InvalidAuthException The grid authToken must be type string on distributed: true');
36+
}
37+
38+
if (this.isMain) {
39+
if (Config.getOption(ConfigOptions.GRID_USE_HTTP))
40+
this.http_s_server = https.createServer({
41+
key: Config.getOption(ConfigOptions.GRID_HTTPS_KEY),
42+
cert: Config.getOption(ConfigOptions.GRID_HTTPS_CERT)
43+
});
44+
else
45+
this.http_s_server = http.createServer();
46+
47+
this.server = new ServerIO.Server(this.http_s_server, {
48+
serveClient: false, // No reason to server bundle client files
49+
connectTimeout: 30 * 1000,
3650
});
37-
else
38-
this.client = ClientIO.io({
39-
// TODO
51+
} else {
52+
let url = Config.getOption(ConfigOptions.GRID_USE_HTTP) ? 'http://' : 'https://';
53+
url += Config.getOption(ConfigOptions.GRID_SERVER_ADDRESS);
54+
55+
const serverPort = Config.getOption(ConfigOptions.GRID_SERVER_ADDRESS);
56+
if (serverPort != null) url += `:${serverPort}`;
57+
58+
this.client = ClientIO.io(url, {
59+
reconnectionDelay: 5 * 1000,
60+
timeout: 15 * 1000,
61+
autoConnect: false,
62+
auth: {
63+
token: Config.getOption(ConfigOptions.GRID_AUTH)
64+
}
4065
});
66+
}
4167
}
4268

4369
/**
4470
* Returns an instance of Grid
4571
*/
4672
static getInstance(): Grid {
4773
if (this.instance == null)
48-
this.instance = new Grid()
74+
this.instance = new Grid();
4975

50-
return this.instance
76+
return this.instance;
5177
}
5278

5379
emit(eventName: string, ...args: any[]): void {
54-
if(!Config.getOption(ConfigOptions.GRID_DISTRIBUTED))
80+
if (!Config.getOption(ConfigOptions.GRID_DISTRIBUTED))
5581
return;
5682

5783
if (this.isMain) {
58-
if(['scheduler.job.push'].includes(eventName))
84+
if (['scheduler.job.push'].includes(eventName))
5985
this.server.emit(eventName, ...args);
60-
}
61-
else {
62-
if(['workers.job.finished', 'workers.job.failed',
63-
'grid.worker.announced', 'grid.worker.destroyed'].includes(eventName))
86+
} else {
87+
if (['workers.job.finished', 'workers.job.failed', 'grid.worker.announced', 'grid.worker.destroyed'].includes(eventName))
6488
this.client.emit(eventName, ...args);
6589
}
6690
}
@@ -72,6 +96,12 @@ export default class Grid {
7296
if (this.isMain) {
7397
if (Config.getOption(ConfigOptions.GRID_DISTRIBUTED)) {
7498
this.server.on("connection", socket => {
99+
const clientAuthToken = socket.handshake.auth.token;
100+
if (!clientAuthToken || clientAuthToken !== Config.getOption(ConfigOptions.GRID_AUTH)) {
101+
socket.disconnect();
102+
return;
103+
}
104+
75105
Events.emit('grid.node.connected', socket);
76106

77107
socket.on("disconnect", () => {
@@ -92,11 +122,12 @@ export default class Grid {
92122

93123
socket.on("grid.worker.destroyed", workerId => {
94124
let index = this.workersIds.findIndex(id => id === workerId);
95-
if (index !== -1) this.workersIds.splice(index, 1);
125+
if (index !== -1)
126+
this.workersIds.splice(index, 1);
96127
});
97128
});
98129

99-
// await this.server.listen(); // TODO - listen with express
130+
this.http_s_server.listen(Config.getOption(ConfigOptions.GRID_SERVER_PORT));
100131
}
101132
} else if (Config.getOption(ConfigOptions.WORKER_NODES) > 0) {
102133
this.client.on('connect', () => {
@@ -109,7 +140,7 @@ export default class Grid {
109140
Events.emit("scheduler.job.push", data);
110141
});
111142

112-
await this.client.connect()
143+
await this.client.connect();
113144
}
114145
}
115146

@@ -173,7 +204,7 @@ export default class Grid {
173204
async finishedJob(job: Job): Promise<void> {
174205
job.status = JobStatus.FINISHED;
175206
if (this.isMain)
176-
Events.emit('workers.job.finished', job)
207+
Events.emit('workers.job.finished', job.id);
177208
else
178209
await this.client.emit('workers.job.finished', job.id);
179210
}
@@ -184,11 +215,11 @@ export default class Grid {
184215
* @param job
185216
*/
186217
async failedJob(job: Job): Promise<void> {
187-
job.status = JobStatus.FAILED
218+
job.status = JobStatus.FAILED;
188219
if (this.isMain)
189-
Events.emit("workers.job.failed", job)
220+
Events.emit("workers.job.failed", job.id);
190221
else
191-
await this.client.emit('workers.job.failed', {id: job.id})
222+
await this.client.emit('workers.job.failed', job.id);
192223
}
193224

194225
/**

0 commit comments

Comments
 (0)
Please sign in to comment.