1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
'use strict';
const common = require('../common');
const assert = require('assert');
const cluster = require('cluster');
const net = require('net');
function forEach(obj, fn) {
Object.keys(obj).forEach(function(name, index) {
fn(obj[name], name);
});
}
if (cluster.isWorker) {
// Create a tcp server. This will be used as cluster-shared-server and as an
// alternative IPC channel.
const server = net.Server();
let socket, message;
function maybeReply() {
if (!socket || !message) return;
// Tell master using TCP socket that a message is received.
socket.write(JSON.stringify({
code: 'received message',
echo: message
}));
}
server.on('connection', function(socket_) {
socket = socket_;
maybeReply();
// Send a message back over the IPC channel.
process.send('message from worker');
});
process.on('message', function(message_) {
message = message_;
maybeReply();
});
server.listen(common.PORT, '127.0.0.1');
} else if (cluster.isMaster) {
const checks = {
global: {
'receive': false,
'correct': false
},
master: {
'receive': false,
'correct': false
},
worker: {
'receive': false,
'correct': false
}
};
let client;
const check = function(type, result) {
checks[type].receive = true;
checks[type].correct = result;
console.error('check', checks);
let missing = false;
forEach(checks, function(type) {
if (type.receive === false) missing = true;
});
if (missing === false) {
console.error('end client');
client.end();
}
};
// Spawn worker
const worker = cluster.fork();
// When a IPC message is received from the worker
worker.on('message', function(message) {
check('master', message === 'message from worker');
});
cluster.on('message', function(worker_, message) {
assert.strictEqual(worker_, worker);
check('global', message === 'message from worker');
});
// When a TCP server is listening in the worker connect to it
worker.on('listening', function() {
client = net.connect(common.PORT, function() {
// Send message to worker.
worker.send('message from master');
});
client.on('data', function(data) {
// All data is JSON
data = JSON.parse(data.toString());
if (data.code === 'received message') {
check('worker', data.echo === 'message from master');
} else {
throw new Error('wrong TCP message received: ' + data);
}
});
// When the connection ends kill worker and shutdown process
client.on('end', function() {
worker.kill();
});
worker.on('exit', function() {
process.exit(0);
});
});
process.once('exit', function() {
forEach(checks, function(check, type) {
assert.ok(check.receive, 'The ' + type + ' did not receive any message');
assert.ok(check.correct,
'The ' + type + ' did not get the correct message');
});
});
}
|
