stream: don't deadlock duplexpair

doc/api/stream.md

@@ -2060,7 +2060,9 @@ when `_read()` is called again after it has stopped should it resume pushing
 additional data onto the queue.
 
 Once the `readable._read()` method has been called, it will not be called again
-until the [`readable.push()`][stream-push] method is called.
+until more data is pushed through the [`readable.push()`][stream-push] method.
+Empty data such as empty buffers and strings will not cause `readable._read()`
+to be called.
 
 The `size` argument is advisory. For implementations where a "read" is a
 single operation that returns data can use the `size` argument to determine how

lib/internal/streams/duplexpair.js

@@ -20,8 +20,12 @@ class DuplexSocket extends Duplex {
   }
 
   _write(chunk, encoding, callback) {
-    this[kOtherSide][kCallback] = callback;
-    this[kOtherSide].push(chunk);
+    if (chunk.length === 0) {
+      process.nextTick(callback);
+    } else {
+      this[kOtherSide].push(chunk);
+      this[kOtherSide][kCallback] = callback;
+    }
   }
 
   _final(callback) {

test/common/duplexpair.js

@@ -24,8 +24,12 @@ class DuplexSocket extends Duplex {
   _write(chunk, encoding, callback) {
     assert.notStrictEqual(this[kOtherSide], null);
     assert.strictEqual(this[kOtherSide][kCallback], null);
-    this[kOtherSide][kCallback] = callback;
-    this[kOtherSide].push(chunk);
+    if (chunk.length === 0) {
+      process.nextTick(callback);
+    } else {
+      this[kOtherSide].push(chunk);
+      this[kOtherSide][kCallback] = callback;
+    }
   }
 
   _final(callback) {

test/parallel/test-http-generic-streams.js

@@ -23,6 +23,7 @@ const MakeDuplexPair = require('../common/duplexpair');
     res.on('data', common.mustCall((data) => {
       assert.strictEqual(data, testData);
     }));
+    res.on('end', common.mustCall());
   }));
   req.end();
 }
@@ -58,3 +59,82 @@ const MakeDuplexPair = require('../common/duplexpair');
     doRequest();
   });
 }
+
+// Test 3: Connection: close request/response with chunked
+{
+  const testData = 'Hello, World!\n';
+  const server = http.createServer(common.mustCall((req, res) => {
+    req.setEncoding('utf8');
+    req.resume();
+    req.on('data', common.mustCall(function test3_req_data(data) {
+      assert.strictEqual(data, testData);
+    }));
+    req.once('end', function() {
+      res.statusCode = 200;
+      res.setHeader('Content-Type', 'text/plain');
+      res.write(testData);
+      res.end();
+    });
+  }));
+
+  const { clientSide, serverSide } = MakeDuplexPair();
+  server.emit('connection', serverSide);
+  clientSide.on('end', common.mustCall());
+  serverSide.on('end', common.mustCall());
+
+  const req = http.request({
+    createConnection: common.mustCall(() => clientSide),
+    method: 'PUT',
+    headers: { 'Connection': 'close' }
+  }, common.mustCall((res) => {
+    res.setEncoding('utf8');
+    res.on('data', common.mustCall(function test3_res_data(data) {
+      assert.strictEqual(data, testData);
+    }));
+    res.on('end', common.mustCall());
+  }));
+  req.write(testData);
+  req.end();
+}
+
+// Test 4: Connection: close request/response with Content-Length
+// The same as Test 3, but with Content-Length headers
+{
+  const testData = 'Hello, World!\n';
+  const server = http.createServer(common.mustCall((req, res) => {
+    assert.strictEqual(req.headers['content-length'], testData.length + '');
+    req.setEncoding('utf8');
+    req.on('data', common.mustCall(function test4_req_data(data) {
+      assert.strictEqual(data, testData);
+    }));
+    req.once('end', function() {
+      res.statusCode = 200;
+      res.setHeader('Content-Type', 'text/plain');
+      res.setHeader('Content-Length', testData.length);
+      res.write(testData);
+      res.end();
+    });
+
+  }));
+
+  const { clientSide, serverSide } = MakeDuplexPair();
+  server.emit('connection', serverSide);
+  clientSide.on('end', common.mustCall());
+  serverSide.on('end', common.mustCall());
+
+  const req = http.request({
+    createConnection: common.mustCall(() => clientSide),
+    method: 'PUT',
+    headers: { 'Connection': 'close' }
+  }, common.mustCall((res) => {
+    res.setEncoding('utf8');
+    assert.strictEqual(res.headers['content-length'], testData.length + '');
+    res.on('data', common.mustCall(function test4_res_data(data) {
+      assert.strictEqual(data, testData);
+    }));
+    res.on('end', common.mustCall());
+  }));
+  req.setHeader('Content-Length', testData.length);
+  req.write(testData);
+  req.end();
+}