http: align with stream.Writable

lib/_http_outgoing.js

@@ -56,7 +56,9 @@ const {
     ERR_METHOD_NOT_IMPLEMENTED,
     ERR_STREAM_CANNOT_PIPE,
     ERR_STREAM_ALREADY_FINISHED,
-    ERR_STREAM_WRITE_AFTER_END
+    ERR_STREAM_WRITE_AFTER_END,
+    ERR_STREAM_NULL_VALUES,
+    ERR_STREAM_DESTROYED
   },
   hideStackFrames
 } = require('internal/errors');
@@ -67,6 +69,8 @@ const { CRLF, debug } = common;
 
 const kCorked = Symbol('corked');
 
+function nop() {}
+
 const RE_CONN_CLOSE = /(?:^|\W)close(?:$|\W)/i;
 const RE_TE_CHUNKED = common.chunkExpression;
 
@@ -641,58 +645,81 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableEnded', {
 
 const crlf_buf = Buffer.from('\r\n');
 OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
+  if (typeof encoding === 'function') {
+    callback = encoding;
+    encoding = null;
+  }
+
   const ret = write_(this, chunk, encoding, callback, false);
   if (!ret)
     this[kNeedDrain] = true;
   return ret;
 };
 
-function writeAfterEnd(msg, callback) {
-  const err = new ERR_STREAM_WRITE_AFTER_END();
+function onError(msg, err, callback) {
   const triggerAsyncId = msg.socket ? msg.socket[async_id_symbol] : undefined;
   defaultTriggerAsyncIdScope(triggerAsyncId,
                              process.nextTick,
-                             writeAfterEndNT,
+                             emitErrorNt,
                              msg,
                              err,
                              callback);
 }
 
+function emitErrorNt(msg, err, callback) {
+  callback(err);
+  if (typeof msg.emit === 'function') msg.emit('error', err);
+}
+
 function write_(msg, chunk, encoding, callback, fromEnd) {
+  if (typeof callback !== 'function')
+    callback = nop;
+
+  let len;
+  if (chunk === null) {
+    throw new ERR_STREAM_NULL_VALUES();
+  } else if (typeof chunk === 'string') {
+    len = Buffer.byteLength(chunk, encoding);
+  } else if (chunk instanceof Buffer) {
+    len = chunk.length;
+  } else {
+    throw new ERR_INVALID_ARG_TYPE(
+      'chunk', ['string', 'Buffer', 'Uint8Array'], chunk);
+  }
+
+  let err;
   if (msg.finished) {
-    writeAfterEnd(msg, callback);
-    return true;
+    err = new ERR_STREAM_WRITE_AFTER_END();
+  } else if (msg.destroyed) {
+    err = new ERR_STREAM_DESTROYED('write');
+  }
+
+  if (err) {
+    onError(msg, err, callback);
+    return false;
   }
 
   if (!msg._header) {
+    if (fromEnd) {
+      msg._contentLength = len;
+    }
     msg._implicitHeader();
   }
 
   if (!msg._hasBody) {
     debug('This type of response MUST NOT have a body. ' +
           'Ignoring write() calls.');
-    if (callback) process.nextTick(callback);
+    process.nextTick(callback);
     return true;
   }
 
-  if (!fromEnd && typeof chunk !== 'string' && !(chunk instanceof Buffer)) {
-    throw new ERR_INVALID_ARG_TYPE('first argument',
-                                   ['string', 'Buffer'], chunk);
-  }
-
   if (!fromEnd && msg.socket && !msg.socket.writableCorked) {
     msg.socket.cork();
     process.nextTick(connectionCorkNT, msg.socket);
   }
 
   let ret;
   if (msg.chunkedEncoding && chunk.length !== 0) {
-    let len;
-    if (typeof chunk === 'string')
-      len = Buffer.byteLength(chunk, encoding);
-    else
-      len = chunk.length;
-
     msg._send(len.toString(16), 'latin1', null);
     msg._send(crlf_buf, null, null);
     msg._send(chunk, encoding, null);
@@ -706,12 +733,6 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
 }
 
 
-function writeAfterEndNT(msg, err, callback) {
-  msg.emit('error', err);
-  if (callback) callback(err);
-}
-
-
 function connectionCorkNT(conn) {
   conn.uncork();
 }
@@ -752,6 +773,7 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
   if (typeof chunk === 'function') {
     callback = chunk;
     chunk = null;
+    encoding = null;
   } else if (typeof encoding === 'function') {
     callback = encoding;
     encoding = null;
@@ -762,21 +784,6 @@ OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
   }
 
   if (chunk) {
-    if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) {
-      throw new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer'], chunk);
-    }
-
-    if (this.finished) {
-      writeAfterEnd(this, callback);
-      return this;
-    }
-
-    if (!this._header) {
-      if (typeof chunk === 'string')
-        this._contentLength = Buffer.byteLength(chunk, encoding);
-      else
-        this._contentLength = chunk.length;
-    }
     write_(this, chunk, encoding, null, true);
   } else if (this.finished) {
     if (typeof callback === 'function') {

test/parallel/test-http-outgoing-destroy.js

@@ -0,0 +1,22 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+
+const http = require('http');
+const OutgoingMessage = http.OutgoingMessage;
+
+{
+  const msg = new OutgoingMessage();
+  assert.strictEqual(msg.destroyed, false);
+  msg.destroy();
+  assert.strictEqual(msg.destroyed, true);
+  let callbackCalled = false;
+  msg.write('asd', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    callbackCalled = true;
+  }));
+  msg.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(callbackCalled, true);
+  }));
+}

test/parallel/test-http-outgoing-finish.js

@@ -20,7 +20,7 @@
 // USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 'use strict';
-require('../common');
+const common = require('../common');
 const assert = require('assert');
 
 const http = require('http');
@@ -49,7 +49,7 @@ function write(out) {
   let endCb = false;
 
   // First, write until it gets some backpressure
-  while (out.write(buf)) {}
+  while (out.write(buf, common.mustCall())) {}
 
   // Now end, and make sure that we don't get the 'finish' event
   // before the tick where the cb gets called.  We give it until
@@ -65,12 +65,12 @@ function write(out) {
     });
   });
 
-  out.end(buf, function() {
+  out.end(buf, common.mustCall(function() {
     endCb = true;
     console.error(`${name} endCb`);
     process.nextTick(function() {
       assert(finishEvent, `${name} got endCb event before finishEvent!`);
       console.log(`ok - ${name} endCb`);
     });
-  });
+  }));
 }

test/parallel/test-http-outgoing-proto.js

@@ -74,16 +74,14 @@ assert.throws(() => {
   );
 }
 
-assert(OutgoingMessage.prototype.write.call({ _header: 'test' }));
-
 assert.throws(() => {
   const outgoingMessage = new OutgoingMessage();
   outgoingMessage.write.call({ _header: 'test', _hasBody: 'test' });
 }, {
   code: 'ERR_INVALID_ARG_TYPE',
   name: 'TypeError',
-  message: 'The first argument must be of type string or an instance of ' +
-           'Buffer. Received undefined'
+  message: 'The "chunk" argument must be of type string or an instance of ' +
+           'Buffer or Uint8Array. Received undefined'
 });
 
 assert.throws(() => {
@@ -92,8 +90,16 @@ assert.throws(() => {
 }, {
   code: 'ERR_INVALID_ARG_TYPE',
   name: 'TypeError',
-  message: 'The first argument must be of type string or an instance of ' +
-           'Buffer. Received type number (1)'
+  message: 'The "chunk" argument must be of type string or an instance of ' +
+           'Buffer or Uint8Array. Received type number (1)'
+});
+
+assert.throws(() => {
+  const outgoingMessage = new OutgoingMessage();
+  outgoingMessage.write.call({ _header: 'test', _hasBody: 'test' }, null);
+}, {
+  code: 'ERR_STREAM_NULL_VALUES',
+  name: 'TypeError'
 });
 
 // addTrailers()

test/parallel/test-http-res-write-after-end.js

@@ -34,8 +34,8 @@ const server = http.Server(common.mustCall(function(req, res) {
   res.end();
 
   const r = res.write('This should raise an error.');
-  // Write after end should return true
-  assert.strictEqual(r, true);
+  // Write after end should return false
+  assert.strictEqual(r, false);
 }));
 
 server.listen(0, function() {