test: Fix many tests for streams2 net refactor

test/simple/test-child-process-disconnect.js

@@ -31,6 +31,8 @@ if (process.argv[2] === 'child') {
 
   server.on('connection', function(socket) {
 
+    socket.resume();
+
     process.on('disconnect', function() {
       socket.end((process.connected).toString());
     });

test/simple/test-child-process-fork-net2.js

@@ -23,31 +23,59 @@ var assert = require('assert');
 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;
   });
 
@@ -61,7 +89,7 @@ if (process.argv[2] === 'child') {
 
   var connected = 0;
   server.on('connection', function(socket) {
-    switch (connected) {
+    switch (connected % 6) {
       case 0:
         child1.send('end', socket); break;
       case 1:
@@ -77,7 +105,7 @@ if (process.argv[2] === 'child') {
     }
     connected += 1;
 
-    if (connected === 6) {
+    if (connected === count) {
       closeServer();
     }
   });
@@ -85,17 +113,23 @@ if (process.argv[2] === 'child') {
   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();
@@ -107,14 +141,18 @@ if (process.argv[2] === 'child') {
 
   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();
@@ -122,8 +160,8 @@ if (process.argv[2] === 'child') {
   };
 
   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');

test/simple/test-cluster-message.js

@@ -81,6 +81,7 @@ else if (cluster.isMaster) {
   var check = function(type, result) {
     checks[type].receive = true;
     checks[type].correct = result;
+    console.error('check', checks);
 
     var missing = false;
     forEach(checks, function(type) {
@@ -88,6 +89,7 @@ else if (cluster.isMaster) {
     });
 
     if (missing === false) {
+      console.error('end client');
       client.end();
     }
   };

test/simple/test-http-1.0-keep-alive.js

@@ -115,6 +115,7 @@ function check(tests) {
   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]);
@@ -126,16 +127,19 @@ function check(tests) {
 
     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);

test/simple/test-net-after-close.js

@@ -25,12 +25,14 @@ var net = require('net');
 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() {

test/simple/test-net-binary.js

@@ -41,12 +41,15 @@ for (var i = 255; i >= 0; i--) {
 
 // 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();
   });
 });
@@ -55,29 +58,44 @@ echoServer.listen(common.PORT);
 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() {

test/simple/test-net-bytes-stats.js

@@ -34,33 +34,40 @@ var count = 0;
 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();

test/simple/test-net-can-reset-timeout.js

@@ -28,6 +28,8 @@ var timeoutCount = 0;
 var server = net.createServer(function(stream) {
   stream.setTimeout(100);
 
+  stream.resume();
+
   stream.on('timeout', function() {
     console.log('timeout');
     // try to reset the timeout.

test/simple/test-net-connect-buffer.js

@@ -38,6 +38,7 @@ var tcp = net.Server(function(s) {
   });
 
   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();
@@ -50,7 +51,7 @@ var tcp = net.Server(function(s) {
 });
 
 tcp.listen(common.PORT, function() {
-  var socket = net.Stream();
+  var socket = net.Stream({ highWaterMark: 0 });
 
   console.log('Connecting to socket ');
 
@@ -77,6 +78,7 @@ tcp.listen(common.PORT, function() {
    {}
   ].forEach(function(v) {
     function f() {
+      console.error('write', v);
       socket.write(v);
     }
     assert.throws(f, TypeError);
@@ -90,12 +92,17 @@ tcp.listen(common.PORT, function() {
   // 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);

test/simple/test-net-connect-options.js

@@ -27,6 +27,7 @@ var serverGotEnd = false;
 var clientGotEnd = false;
 
 var server = net.createServer({allowHalfOpen: true}, function(socket) {
+  socket.resume();
   socket.on('end', function() {
     serverGotEnd = true;
   });
@@ -39,6 +40,8 @@ server.listen(common.PORT, function() {
     port: common.PORT,
     allowHalfOpen: true
   }, function() {
+    console.error('client connect cb');
+    client.resume();
     client.on('end', function() {
       clientGotEnd = true;
       setTimeout(function() {
@@ -53,6 +56,7 @@ server.listen(common.PORT, function() {
 });
 
 process.on('exit', function() {
+  console.error('exit', serverGotEnd, clientGotEnd);
   assert(serverGotEnd);
   assert(clientGotEnd);
 });

test/simple/test-net-pingpong.js

@@ -60,6 +60,8 @@ function pingPongTest(port, host) {
     });
 
     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();
@@ -129,10 +131,11 @@ function pingPongTest(port, host) {
 }
 
 /* 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);

test/simple/test-net-reconnect.js

@@ -30,39 +30,49 @@ var client_recv_count = 0;
 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

test/simple/test-net-remote-address-port.js

@@ -34,6 +34,7 @@ var server = net.createServer(function(socket) {
   socket.on('end', function() {
     if (++conns_closed == 2) server.close();
   });
+  socket.resume();
 });
 
 server.listen(common.PORT, 'localhost', function() {

test/simple/test-net-write-after-close.js

@@ -32,12 +32,16 @@ process.on('exit', 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;
     });

test/simple/test-pipe-file-to-http.js

@@ -31,6 +31,7 @@ var clientReqComplete = false;
 var count = 0;
 
 var server = http.createServer(function(req, res) {
+  console.error('SERVER request');
   var timeoutId;
   assert.equal('POST', req.method);
   req.pause();
@@ -63,6 +64,8 @@ server.on('listening', function() {
 
   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();
   });
 });
@@ -75,8 +78,15 @@ function 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;
@@ -84,7 +94,10 @@ function makeRequest() {
   });
 
   req.on('response', function(res) {
+    console.error('RESPONSE', res.statusCode, res.headers);
+    res.resume();
     res.on('end', function() {
+      console.error('RESPONSE end');
       server.close();
     });
   });

test/simple/test-pipe.js

@@ -125,6 +125,7 @@ function startClient() {
   });
   req.write(buffer);
   req.end();
+  console.error('ended request', req);
 }
 
 process.on('exit', function() {

test/simple/test-tcp-wrap-connect.js

@@ -54,6 +54,7 @@ var shutdownCount = 0;
 var server = require('net').Server(function(s) {
   console.log('got connection');
   connectCount++;
+  s.resume();
   s.on('end', function() {
     console.log('got eof');
     endCount++;

test/simple/test-zlib-random-byte-pipes.js

@@ -150,8 +150,25 @@ var inp = new RandomReadStream({ total: 1024, block: 256, jitter: 16 });
 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;