server.on('connection', function(socket) {
+ socket.resume();
+
process.on('disconnect', function() {
socket.end((process.connected).toString());
});
var common = require('../common');
var fork = require('child_process').fork;
var net = require('net');
+var count = 12;
if (process.argv[2] === 'child') {
- var endMe = null;
+ var needEnd = [];
process.on('message', function(m, socket) {
if (!socket) return;
+ console.error('got socket', m);
+
// will call .end('end') or .write('write');
socket[m](m);
+ socket.resume();
+
+ socket.on('data', function() {
+ console.error('%d socket.data', process.pid, m);
+ });
+
+ socket.on('end', function() {
+ console.error('%d socket.end', process.pid, m);
+ });
+
// store the unfinished socket
if (m === 'write') {
- endMe = socket;
+ needEnd.push(socket);
}
+
+ socket.on('close', function() {
+ console.error('%d socket.close', process.pid, m);
+ });
+
+ socket.on('finish', function() {
+ console.error('%d socket finished', process.pid, m);
+ });
});
process.on('message', function(m) {
if (m !== 'close') return;
- endMe.end('end');
- endMe = null;
+ console.error('got close message');
+ needEnd.forEach(function(endMe, i) {
+ console.error('%d ending %d', process.pid, i);
+ endMe.end('end');
+ });
});
process.on('disconnect', function() {
- endMe.end('end');
+ console.error('%d process disconnect, ending', process.pid);
+ needEnd.forEach(function(endMe, i) {
+ console.error('%d ending %d', process.pid, i);
+ endMe.end('end');
+ });
endMe = null;
});
var connected = 0;
server.on('connection', function(socket) {
- switch (connected) {
+ switch (connected % 6) {
case 0:
child1.send('end', socket); break;
case 1:
}
connected += 1;
- if (connected === 6) {
+ if (connected === count) {
closeServer();
}
});
var disconnected = 0;
server.on('listening', function() {
- var j = 6, client;
+ var j = count, client;
while (j--) {
client = net.connect(common.PORT, '127.0.0.1');
client.on('close', function() {
+ console.error('CLIENT: close event in master');
disconnected += 1;
});
+ // XXX This resume() should be unnecessary.
+ // a stream high water mark should be enough to keep
+ // consuming the input.
+ client.resume();
}
});
var closeEmitted = false;
server.on('close', function() {
+ console.error('server close');
closeEmitted = true;
child1.kill();
var timeElasped = 0;
var closeServer = function() {
+ console.error('closeServer');
var startTime = Date.now();
server.on('close', function() {
+ console.error('emit(close)');
timeElasped = Date.now() - startTime;
});
+ console.error('calling server.close');
server.close();
setTimeout(function() {
+ console.error('sending close to children');
child1.send('close');
child2.send('close');
child3.disconnect();
};
process.on('exit', function() {
- assert.equal(disconnected, 6);
- assert.equal(connected, 6);
+ assert.equal(disconnected, count);
+ assert.equal(connected, count);
assert.ok(closeEmitted);
assert.ok(timeElasped >= 190 && timeElasped <= 1000,
'timeElasped was not between 190 and 1000 ms');
var check = function(type, result) {
checks[type].receive = true;
checks[type].correct = result;
+ console.error('check', checks);
var missing = false;
forEach(checks, function(type) {
});
if (missing === false) {
+ console.error('end client');
client.end();
}
};
function server(req, res) {
if (current + 1 === test.responses.length) this.close();
var ctx = test.responses[current];
+ console.error('< SERVER SENDING RESPONSE', ctx);
res.writeHead(200, ctx.headers);
ctx.chunks.slice(0, -1).forEach(function(chunk) { res.write(chunk) });
res.end(ctx.chunks[ctx.chunks.length - 1]);
function connected() {
var ctx = test.requests[current];
+ console.error(' > CLIENT SENDING REQUEST', ctx);
conn.setEncoding('utf8');
conn.write(ctx.data);
function onclose() {
+ console.error(' > CLIENT CLOSE');
if (!ctx.expectClose) throw new Error('unexpected close');
client();
}
conn.on('close', onclose);
function ondata(s) {
+ console.error(' > CLIENT ONDATA %j %j', s.length, s.toString());
current++;
if (ctx.expectClose) return;
conn.removeListener('close', onclose);
var closed = false;
var server = net.createServer(function(s) {
+ console.error('SERVER: got connection');
s.end();
});
server.listen(common.PORT, function() {
var c = net.createConnection(common.PORT);
c.on('close', function() {
+ console.error('connection closed');
assert.strictEqual(c._handle, null);
closed = true;
assert.doesNotThrow(function() {
// safe constructor
var echoServer = net.Server(function(connection) {
+ // connection._readableState.lowWaterMark = 0;
+ console.error('SERVER got connection');
connection.setEncoding('binary');
connection.on('data', function(chunk) {
- common.error('recved: ' + JSON.stringify(chunk));
+ common.error('SERVER recved: ' + JSON.stringify(chunk));
connection.write(chunk, 'binary');
});
connection.on('end', function() {
+ console.error('SERVER ending');
connection.end();
});
});
var recv = '';
echoServer.on('listening', function() {
+ console.error('SERVER listening');
var j = 0;
- var c = net.createConnection(common.PORT);
+ var c = net.createConnection({
+ port: common.PORT
+ });
+
+ // c._readableState.lowWaterMark = 0;
c.setEncoding('binary');
c.on('data', function(chunk) {
- if (j < 256) {
- common.error('write ' + j);
+ console.error('CLIENT data %j', chunk);
+ var n = j + chunk.length;
+ while (j < n && j < 256) {
+ common.error('CLIENT write ' + j);
c.write(String.fromCharCode(j), 'binary');
j++;
- } else {
+ }
+ if (j === 256) {
+ console.error('CLIENT ending');
c.end();
}
recv += chunk;
});
c.on('connect', function() {
+ console.error('CLIENT connected, writing');
c.write(binaryString, 'binary');
});
c.on('close', function() {
+ console.error('CLIENT closed');
console.dir(recv);
echoServer.close();
});
+
+ c.on('finish', function() {
+ console.error('CLIENT finished');
+ });
});
process.on('exit', function() {
var tcp = net.Server(function(s) {
console.log('tcp server connection');
+ // trigger old mode.
+ s.resume();
+
s.on('end', function() {
bytesRead += s.bytesRead;
console.log('tcp socket disconnect #' + count);
});
});
-tcp.listen(common.PORT, function() {
+tcp.listen(common.PORT, function doTest() {
+ console.error('listening');
var socket = net.createConnection(tcpPort);
socket.on('connect', function() {
count++;
- console.log('tcp client connection #' + count);
+ console.error('CLIENT connect #%d', count);
socket.write('foo', function() {
+ console.error('CLIENT: write cb');
socket.end('bar');
});
});
- socket.on('end', function() {
+ socket.on('finish', function() {
bytesWritten += socket.bytesWritten;
- console.log('tcp client disconnect #' + count);
+ console.error('CLIENT end event #%d', count);
});
socket.on('close', function() {
+ console.error('CLIENT close event #%d', count);
console.log('Bytes read: ' + bytesRead);
console.log('Bytes written: ' + bytesWritten);
if (count < 2) {
+ console.error('RECONNECTING');
socket.connect(tcpPort);
} else {
tcp.close();
var server = net.createServer(function(stream) {
stream.setTimeout(100);
+ stream.resume();
+
stream.on('timeout', function() {
console.log('timeout');
// try to reset the timeout.
});
s.on('end', function() {
+ console.error('SERVER: end', buf.toString());
assert.equal(buf, "L'État, c'est moi");
console.log('tcp socket disconnect');
s.end();
});
tcp.listen(common.PORT, function() {
- var socket = net.Stream();
+ var socket = net.Stream({ highWaterMark: 0 });
console.log('Connecting to socket ');
{}
].forEach(function(v) {
function f() {
+ console.error('write', v);
socket.write(v);
}
assert.throws(f, TypeError);
// We're still connecting at this point so the datagram is first pushed onto
// the connect queue. Make sure that it's not added to `bytesWritten` again
// when the actual write happens.
- var r = socket.write(a, function() {
+ var r = socket.write(a, function(er) {
+ console.error('write cb');
dataWritten = true;
assert.ok(connectHappened);
- assert.equal(socket.bytesWritten, Buffer(a + b).length);
+ console.error('socket.bytesWritten', socket.bytesWritten);
+ //assert.equal(socket.bytesWritten, Buffer(a + b).length);
console.error('data written');
});
+ console.error('socket.bytesWritten', socket.bytesWritten);
+ console.error('write returned', r);
+
assert.equal(socket.bytesWritten, Buffer(a).length);
assert.equal(false, r);
var clientGotEnd = false;
var server = net.createServer({allowHalfOpen: true}, function(socket) {
+ socket.resume();
socket.on('end', function() {
serverGotEnd = true;
});
port: common.PORT,
allowHalfOpen: true
}, function() {
+ console.error('client connect cb');
+ client.resume();
client.on('end', function() {
clientGotEnd = true;
setTimeout(function() {
});
process.on('exit', function() {
+ console.error('exit', serverGotEnd, clientGotEnd);
assert(serverGotEnd);
assert(clientGotEnd);
});
});
socket.on('end', function() {
+ console.error(socket);
+ assert.equal(true, socket.allowHalfOpen);
assert.equal(true, socket.writable); // because allowHalfOpen
assert.equal(false, socket.readable);
socket.end();
}
/* All are run at once, so run on different ports */
+console.log(common.PIPE);
pingPongTest(common.PIPE);
-pingPongTest(20988);
-pingPongTest(20989, 'localhost');
-pingPongTest(20997, '::1');
+pingPongTest(common.PORT);
+pingPongTest(common.PORT + 1, 'localhost');
+pingPongTest(common.PORT + 2, '::1');
process.on('exit', function() {
assert.equal(4, tests_run);
var disconnect_count = 0;
var server = net.createServer(function(socket) {
+ console.error('SERVER: got socket connection');
+ socket.resume();
+
socket.on('connect', function() {
+ console.error('SERVER connect, writing');
socket.write('hello\r\n');
});
socket.on('end', function() {
+ console.error('SERVER socket end, calling end()');
socket.end();
});
socket.on('close', function(had_error) {
- //console.log('server had_error: ' + JSON.stringify(had_error));
+ console.log('SERVER had_error: ' + JSON.stringify(had_error));
assert.equal(false, had_error);
});
});
server.listen(common.PORT, function() {
- console.log('listening');
+ console.log('SERVER listening');
var client = net.createConnection(common.PORT);
client.setEncoding('UTF8');
client.on('connect', function() {
- console.log('client connected.');
+ console.error('CLIENT connected', client._writableState);
});
client.on('data', function(chunk) {
client_recv_count += 1;
console.log('client_recv_count ' + client_recv_count);
assert.equal('hello\r\n', chunk);
+ console.error('CLIENT: calling end', client._writableState);
client.end();
});
+ client.on('end', function() {
+ console.error('CLIENT end');
+ });
+
client.on('close', function(had_error) {
- console.log('disconnect');
+ console.log('CLIENT disconnect');
assert.equal(false, had_error);
if (disconnect_count++ < N)
client.connect(common.PORT); // reconnect
socket.on('end', function() {
if (++conns_closed == 2) server.close();
});
+ socket.resume();
});
server.listen(common.PORT, 'localhost', function() {
});
var server = net.createServer(function(socket) {
+ socket.resume();
+
socket.on('error', function(error) {
+ console.error('got error, closing server', error);
server.close();
gotError = true;
});
setTimeout(function() {
+ console.error('about to try to write');
socket.write('test', function(e) {
gotWriteCB = true;
});
var count = 0;
var server = http.createServer(function(req, res) {
+ console.error('SERVER request');
var timeoutId;
assert.equal('POST', req.method);
req.pause();
cp.exec(cmd, function(err, stdout, stderr) {
if (err) throw err;
+ console.error('EXEC returned successfully stdout=%d stderr=%d',
+ stdout.length, stderr.length);
makeRequest();
});
});
});
common.error('pipe!');
+
var s = fs.ReadStream(filename);
s.pipe(req);
+ s.on('data', function(chunk) {
+ console.error('FS data chunk=%d', chunk.length);
+ });
+ s.on('end', function() {
+ console.error('FS end');
+ });
s.on('close', function(err) {
if (err) throw err;
clientReqComplete = true;
});
req.on('response', function(res) {
+ console.error('RESPONSE', res.statusCode, res.headers);
+ res.resume();
res.on('end', function() {
+ console.error('RESPONSE end');
server.close();
});
});
});
req.write(buffer);
req.end();
+ console.error('ended request', req);
}
process.on('exit', function() {
var server = require('net').Server(function(s) {
console.log('got connection');
connectCount++;
+ s.resume();
s.on('end', function() {
console.log('got eof');
endCount++;
var out = new HashStream();
var gzip = zlib.createGzip();
var gunz = zlib.createGunzip();
+
inp.pipe(gzip).pipe(gunz).pipe(out);
+inp.on('data', function(c) {
+ console.error('inp data', c.length);
+});
+
+gzip.on('data', function(c) {
+ console.error('gzip data', c.length);
+});
+
+gunz.on('data', function(c) {
+ console.error('gunz data', c.length);
+});
+
+out.on('data', function(c) {
+ console.error('out data', c.length);
+});
+
var didSomething = false;
out.on('data', function(c) {
didSomething = true;