mirror of
https://github.com/nodejs/node.git
synced 2024-11-29 23:16:30 +01:00
db706da235
Invoke callback with ERR_STREAM_ALREADY_FINISHED error if `end()` is called on a finished stream. PR-URL: https://github.com/nodejs/node/pull/28687 Refs: https://github.com/nodejs/node/issues/28667 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com>
855 lines
24 KiB
JavaScript
855 lines
24 KiB
JavaScript
// Copyright Joyent, Inc. and other Node contributors.
|
|
//
|
|
// Permission is hereby granted, free of charge, to any person obtaining a
|
|
// copy of this software and associated documentation files (the
|
|
// "Software"), to deal in the Software without restriction, including
|
|
// without limitation the rights to use, copy, modify, merge, publish,
|
|
// distribute, sublicense, and/or sell copies of the Software, and to permit
|
|
// persons to whom the Software is furnished to do so, subject to the
|
|
// following conditions:
|
|
//
|
|
// The above copyright notice and this permission notice shall be included
|
|
// in all copies or substantial portions of the Software.
|
|
//
|
|
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
|
|
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
|
|
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
|
|
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
|
|
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
|
|
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
|
|
// USE OR OTHER DEALINGS IN THE SOFTWARE.
|
|
|
|
'use strict';
|
|
|
|
const { Object, ObjectPrototype } = primordials;
|
|
|
|
const { getDefaultHighWaterMark } = require('internal/streams/state');
|
|
const assert = require('internal/assert');
|
|
const Stream = require('stream');
|
|
const internalUtil = require('internal/util');
|
|
const { kOutHeaders, utcDate, kNeedDrain } = require('internal/http');
|
|
const { Buffer } = require('buffer');
|
|
const common = require('_http_common');
|
|
const checkIsHttpToken = common._checkIsHttpToken;
|
|
const checkInvalidHeaderChar = common._checkInvalidHeaderChar;
|
|
const {
|
|
defaultTriggerAsyncIdScope,
|
|
symbols: { async_id_symbol }
|
|
} = require('internal/async_hooks');
|
|
const {
|
|
codes: {
|
|
ERR_HTTP_HEADERS_SENT,
|
|
ERR_HTTP_INVALID_HEADER_VALUE,
|
|
ERR_HTTP_TRAILER_INVALID,
|
|
ERR_INVALID_HTTP_TOKEN,
|
|
ERR_INVALID_ARG_TYPE,
|
|
ERR_INVALID_CHAR,
|
|
ERR_METHOD_NOT_IMPLEMENTED,
|
|
ERR_STREAM_CANNOT_PIPE,
|
|
ERR_STREAM_ALREADY_FINISHED,
|
|
ERR_STREAM_WRITE_AFTER_END
|
|
},
|
|
hideStackFrames
|
|
} = require('internal/errors');
|
|
const { validateString } = require('internal/validators');
|
|
|
|
const HIGH_WATER_MARK = getDefaultHighWaterMark();
|
|
const { CRLF, debug } = common;
|
|
|
|
const kIsCorked = Symbol('isCorked');
|
|
|
|
const RE_CONN_CLOSE = /(?:^|\W)close(?:$|\W)/i;
|
|
const RE_TE_CHUNKED = common.chunkExpression;
|
|
|
|
// isCookieField performs a case-insensitive comparison of a provided string
|
|
// against the word "cookie." As of V8 6.6 this is faster than handrolling or
|
|
// using a case-insensitive RegExp.
|
|
function isCookieField(s) {
|
|
return s.length === 6 && s.toLowerCase() === 'cookie';
|
|
}
|
|
|
|
function noopPendingOutput(amount) {}
|
|
|
|
function OutgoingMessage() {
|
|
Stream.call(this);
|
|
|
|
// Queue that holds all currently pending data, until the response will be
|
|
// assigned to the socket (until it will its turn in the HTTP pipeline).
|
|
this.outputData = [];
|
|
|
|
// `outputSize` is an approximate measure of how much data is queued on this
|
|
// response. `_onPendingData` will be invoked to update similar global
|
|
// per-connection counter. That counter will be used to pause/unpause the
|
|
// TCP socket and HTTP Parser and thus handle the backpressure.
|
|
this.outputSize = 0;
|
|
|
|
this.writable = true;
|
|
|
|
this._last = false;
|
|
this.chunkedEncoding = false;
|
|
this.shouldKeepAlive = true;
|
|
this.useChunkedEncodingByDefault = true;
|
|
this.sendDate = false;
|
|
this._removedConnection = false;
|
|
this._removedContLen = false;
|
|
this._removedTE = false;
|
|
|
|
this._contentLength = null;
|
|
this._hasBody = true;
|
|
this._trailer = '';
|
|
this[kNeedDrain] = false;
|
|
|
|
this.finished = false;
|
|
this._headerSent = false;
|
|
this[kIsCorked] = false;
|
|
|
|
this.socket = null;
|
|
this._header = null;
|
|
this[kOutHeaders] = null;
|
|
|
|
this._onPendingData = noopPendingOutput;
|
|
}
|
|
Object.setPrototypeOf(OutgoingMessage.prototype, Stream.prototype);
|
|
Object.setPrototypeOf(OutgoingMessage, Stream);
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'writableFinished', {
|
|
get() {
|
|
return (
|
|
this.finished &&
|
|
this.outputSize === 0 &&
|
|
(!this.socket || this.socket.writableLength === 0)
|
|
);
|
|
}
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'writableObjectMode', {
|
|
get() {
|
|
return false;
|
|
}
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'writableLength', {
|
|
get() {
|
|
return this.outputSize + (this.socket ? this.socket.writableLength : 0);
|
|
}
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'writableHighWaterMark', {
|
|
get() {
|
|
return this.socket ? this.socket.writableHighWaterMark : HIGH_WATER_MARK;
|
|
}
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, '_headers', {
|
|
get: internalUtil.deprecate(function() {
|
|
return this.getHeaders();
|
|
}, 'OutgoingMessage.prototype._headers is deprecated', 'DEP0066'),
|
|
set: internalUtil.deprecate(function(val) {
|
|
if (val == null) {
|
|
this[kOutHeaders] = null;
|
|
} else if (typeof val === 'object') {
|
|
const headers = this[kOutHeaders] = Object.create(null);
|
|
const keys = Object.keys(val);
|
|
for (var i = 0; i < keys.length; ++i) {
|
|
const name = keys[i];
|
|
headers[name.toLowerCase()] = [name, val[name]];
|
|
}
|
|
}
|
|
}, 'OutgoingMessage.prototype._headers is deprecated', 'DEP0066')
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'connection', {
|
|
get: function() {
|
|
return this.socket;
|
|
},
|
|
set: function(val) {
|
|
this.socket = val;
|
|
}
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, '_headerNames', {
|
|
get: internalUtil.deprecate(function() {
|
|
const headers = this[kOutHeaders];
|
|
if (headers !== null) {
|
|
const out = Object.create(null);
|
|
const keys = Object.keys(headers);
|
|
for (var i = 0; i < keys.length; ++i) {
|
|
const key = keys[i];
|
|
const val = headers[key][0];
|
|
out[key] = val;
|
|
}
|
|
return out;
|
|
}
|
|
return null;
|
|
}, 'OutgoingMessage.prototype._headerNames is deprecated', 'DEP0066'),
|
|
set: internalUtil.deprecate(function(val) {
|
|
if (typeof val === 'object' && val !== null) {
|
|
const headers = this[kOutHeaders];
|
|
if (!headers)
|
|
return;
|
|
const keys = Object.keys(val);
|
|
for (var i = 0; i < keys.length; ++i) {
|
|
const header = headers[keys[i]];
|
|
if (header)
|
|
header[0] = val[keys[i]];
|
|
}
|
|
}
|
|
}, 'OutgoingMessage.prototype._headerNames is deprecated', 'DEP0066')
|
|
});
|
|
|
|
|
|
OutgoingMessage.prototype._renderHeaders = function _renderHeaders() {
|
|
if (this._header) {
|
|
throw new ERR_HTTP_HEADERS_SENT('render');
|
|
}
|
|
|
|
const headersMap = this[kOutHeaders];
|
|
const headers = {};
|
|
|
|
if (headersMap !== null) {
|
|
const keys = Object.keys(headersMap);
|
|
for (var i = 0, l = keys.length; i < l; i++) {
|
|
const key = keys[i];
|
|
headers[headersMap[key][0]] = headersMap[key][1];
|
|
}
|
|
}
|
|
return headers;
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype.setTimeout = function setTimeout(msecs, callback) {
|
|
|
|
if (callback) {
|
|
this.on('timeout', callback);
|
|
}
|
|
|
|
if (!this.socket) {
|
|
this.once('socket', function socketSetTimeoutOnConnect(socket) {
|
|
socket.setTimeout(msecs);
|
|
});
|
|
} else {
|
|
this.socket.setTimeout(msecs);
|
|
}
|
|
return this;
|
|
};
|
|
|
|
|
|
// It's possible that the socket will be destroyed, and removed from
|
|
// any messages, before ever calling this. In that case, just skip
|
|
// it, since something else is destroying this connection anyway.
|
|
OutgoingMessage.prototype.destroy = function destroy(error) {
|
|
if (this.socket) {
|
|
this.socket.destroy(error);
|
|
} else {
|
|
this.once('socket', function socketDestroyOnConnect(socket) {
|
|
socket.destroy(error);
|
|
});
|
|
}
|
|
};
|
|
|
|
|
|
// This abstract either writing directly to the socket or buffering it.
|
|
OutgoingMessage.prototype._send = function _send(data, encoding, callback) {
|
|
// This is a shameful hack to get the headers and first body chunk onto
|
|
// the same packet. Future versions of Node are going to take care of
|
|
// this at a lower level and in a more general way.
|
|
if (!this._headerSent) {
|
|
if (typeof data === 'string' &&
|
|
(encoding === 'utf8' || encoding === 'latin1' || !encoding)) {
|
|
data = this._header + data;
|
|
} else {
|
|
var header = this._header;
|
|
if (this.outputData.length === 0) {
|
|
this.outputData = [{
|
|
data: header,
|
|
encoding: 'latin1',
|
|
callback: null
|
|
}];
|
|
} else {
|
|
this.outputData.unshift({
|
|
data: header,
|
|
encoding: 'latin1',
|
|
callback: null
|
|
});
|
|
}
|
|
this.outputSize += header.length;
|
|
this._onPendingData(header.length);
|
|
}
|
|
this._headerSent = true;
|
|
}
|
|
return this._writeRaw(data, encoding, callback);
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype._writeRaw = _writeRaw;
|
|
function _writeRaw(data, encoding, callback) {
|
|
const conn = this.socket;
|
|
if (conn && conn.destroyed) {
|
|
// The socket was destroyed. If we're still trying to write to it,
|
|
// then we haven't gotten the 'close' event yet.
|
|
return false;
|
|
}
|
|
|
|
if (typeof encoding === 'function') {
|
|
callback = encoding;
|
|
encoding = null;
|
|
}
|
|
|
|
if (conn && conn._httpMessage === this && conn.writable) {
|
|
// There might be pending data in the this.output buffer.
|
|
if (this.outputData.length) {
|
|
this._flushOutput(conn);
|
|
}
|
|
// Directly write to socket.
|
|
return conn.write(data, encoding, callback);
|
|
}
|
|
// Buffer, as long as we're not destroyed.
|
|
this.outputData.push({ data, encoding, callback });
|
|
this.outputSize += data.length;
|
|
this._onPendingData(data.length);
|
|
return this.outputSize < HIGH_WATER_MARK;
|
|
}
|
|
|
|
|
|
OutgoingMessage.prototype._storeHeader = _storeHeader;
|
|
function _storeHeader(firstLine, headers) {
|
|
// firstLine in the case of request is: 'GET /index.html HTTP/1.1\r\n'
|
|
// in the case of response it is: 'HTTP/1.1 200 OK\r\n'
|
|
const state = {
|
|
connection: false,
|
|
contLen: false,
|
|
te: false,
|
|
date: false,
|
|
expect: false,
|
|
trailer: false,
|
|
header: firstLine
|
|
};
|
|
|
|
if (headers) {
|
|
if (headers === this[kOutHeaders]) {
|
|
for (const key in headers) {
|
|
const entry = headers[key];
|
|
processHeader(this, state, entry[0], entry[1], false);
|
|
}
|
|
} else if (Array.isArray(headers)) {
|
|
for (const entry of headers) {
|
|
processHeader(this, state, entry[0], entry[1], true);
|
|
}
|
|
} else {
|
|
for (const key in headers) {
|
|
if (ObjectPrototype.hasOwnProperty(headers, key)) {
|
|
processHeader(this, state, key, headers[key], true);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let { header } = state;
|
|
|
|
// Date header
|
|
if (this.sendDate && !state.date) {
|
|
header += 'Date: ' + utcDate() + CRLF;
|
|
}
|
|
|
|
// Force the connection to close when the response is a 204 No Content or
|
|
// a 304 Not Modified and the user has set a "Transfer-Encoding: chunked"
|
|
// header.
|
|
//
|
|
// RFC 2616 mandates that 204 and 304 responses MUST NOT have a body but
|
|
// node.js used to send out a zero chunk anyway to accommodate clients
|
|
// that don't have special handling for those responses.
|
|
//
|
|
// It was pointed out that this might confuse reverse proxies to the point
|
|
// of creating security liabilities, so suppress the zero chunk and force
|
|
// the connection to close.
|
|
if (this.chunkedEncoding && (this.statusCode === 204 ||
|
|
this.statusCode === 304)) {
|
|
debug(this.statusCode + ' response should not use chunked encoding,' +
|
|
' closing connection.');
|
|
this.chunkedEncoding = false;
|
|
this.shouldKeepAlive = false;
|
|
}
|
|
|
|
// keep-alive logic
|
|
if (this._removedConnection) {
|
|
this._last = true;
|
|
this.shouldKeepAlive = false;
|
|
} else if (!state.connection) {
|
|
const shouldSendKeepAlive = this.shouldKeepAlive &&
|
|
(state.contLen || this.useChunkedEncodingByDefault || this.agent);
|
|
if (shouldSendKeepAlive) {
|
|
header += 'Connection: keep-alive\r\n';
|
|
} else {
|
|
this._last = true;
|
|
header += 'Connection: close\r\n';
|
|
}
|
|
}
|
|
|
|
if (!state.contLen && !state.te) {
|
|
if (!this._hasBody) {
|
|
// Make sure we don't end the 0\r\n\r\n at the end of the message.
|
|
this.chunkedEncoding = false;
|
|
} else if (!this.useChunkedEncodingByDefault) {
|
|
this._last = true;
|
|
} else if (!state.trailer &&
|
|
!this._removedContLen &&
|
|
typeof this._contentLength === 'number') {
|
|
header += 'Content-Length: ' + this._contentLength + CRLF;
|
|
} else if (!this._removedTE) {
|
|
header += 'Transfer-Encoding: chunked\r\n';
|
|
this.chunkedEncoding = true;
|
|
} else {
|
|
// We should only be able to get here if both Content-Length and
|
|
// Transfer-Encoding are removed by the user.
|
|
// See: test/parallel/test-http-remove-header-stays-removed.js
|
|
debug('Both Content-Length and Transfer-Encoding are removed');
|
|
}
|
|
}
|
|
|
|
// Test non-chunked message does not have trailer header set,
|
|
// message will be terminated by the first empty line after the
|
|
// header fields, regardless of the header fields present in the
|
|
// message, and thus cannot contain a message body or 'trailers'.
|
|
if (this.chunkedEncoding !== true && state.trailer) {
|
|
throw new ERR_HTTP_TRAILER_INVALID();
|
|
}
|
|
|
|
this._header = header + CRLF;
|
|
this._headerSent = false;
|
|
|
|
// Wait until the first body chunk, or close(), is sent to flush,
|
|
// UNLESS we're sending Expect: 100-continue.
|
|
if (state.expect) this._send('');
|
|
}
|
|
|
|
function processHeader(self, state, key, value, validate) {
|
|
if (validate)
|
|
validateHeaderName(key);
|
|
if (Array.isArray(value)) {
|
|
if (value.length < 2 || !isCookieField(key)) {
|
|
for (var i = 0; i < value.length; i++)
|
|
storeHeader(self, state, key, value[i], validate);
|
|
return;
|
|
}
|
|
value = value.join('; ');
|
|
}
|
|
storeHeader(self, state, key, value, validate);
|
|
}
|
|
|
|
function storeHeader(self, state, key, value, validate) {
|
|
if (validate)
|
|
validateHeaderValue(key, value);
|
|
state.header += key + ': ' + value + CRLF;
|
|
matchHeader(self, state, key, value);
|
|
}
|
|
|
|
function matchHeader(self, state, field, value) {
|
|
if (field.length < 4 || field.length > 17)
|
|
return;
|
|
field = field.toLowerCase();
|
|
switch (field) {
|
|
case 'connection':
|
|
state.connection = true;
|
|
self._removedConnection = false;
|
|
if (RE_CONN_CLOSE.test(value))
|
|
self._last = true;
|
|
else
|
|
self.shouldKeepAlive = true;
|
|
break;
|
|
case 'transfer-encoding':
|
|
state.te = true;
|
|
self._removedTE = false;
|
|
if (RE_TE_CHUNKED.test(value)) self.chunkedEncoding = true;
|
|
break;
|
|
case 'content-length':
|
|
state.contLen = true;
|
|
self._removedContLen = false;
|
|
break;
|
|
case 'date':
|
|
case 'expect':
|
|
case 'trailer':
|
|
state[field] = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
const validateHeaderName = hideStackFrames((name) => {
|
|
if (typeof name !== 'string' || !name || !checkIsHttpToken(name)) {
|
|
throw new ERR_INVALID_HTTP_TOKEN('Header name', name);
|
|
}
|
|
});
|
|
|
|
const validateHeaderValue = hideStackFrames((name, value) => {
|
|
if (value === undefined) {
|
|
throw new ERR_HTTP_INVALID_HEADER_VALUE(value, name);
|
|
}
|
|
if (checkInvalidHeaderChar(value)) {
|
|
debug('Header "%s" contains invalid characters', name);
|
|
throw new ERR_INVALID_CHAR('header content', name);
|
|
}
|
|
});
|
|
|
|
OutgoingMessage.prototype.setHeader = function setHeader(name, value) {
|
|
if (this._header) {
|
|
throw new ERR_HTTP_HEADERS_SENT('set');
|
|
}
|
|
validateHeaderName(name);
|
|
validateHeaderValue(name, value);
|
|
|
|
let headers = this[kOutHeaders];
|
|
if (headers === null)
|
|
this[kOutHeaders] = headers = Object.create(null);
|
|
|
|
headers[name.toLowerCase()] = [name, value];
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype.getHeader = function getHeader(name) {
|
|
validateString(name, 'name');
|
|
|
|
const headers = this[kOutHeaders];
|
|
if (headers === null)
|
|
return;
|
|
|
|
const entry = headers[name.toLowerCase()];
|
|
return entry && entry[1];
|
|
};
|
|
|
|
|
|
// Returns an array of the names of the current outgoing headers.
|
|
OutgoingMessage.prototype.getHeaderNames = function getHeaderNames() {
|
|
return this[kOutHeaders] !== null ? Object.keys(this[kOutHeaders]) : [];
|
|
};
|
|
|
|
|
|
// Returns a shallow copy of the current outgoing headers.
|
|
OutgoingMessage.prototype.getHeaders = function getHeaders() {
|
|
const headers = this[kOutHeaders];
|
|
const ret = Object.create(null);
|
|
if (headers) {
|
|
const keys = Object.keys(headers);
|
|
for (var i = 0; i < keys.length; ++i) {
|
|
const key = keys[i];
|
|
const val = headers[key][1];
|
|
ret[key] = val;
|
|
}
|
|
}
|
|
return ret;
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype.hasHeader = function hasHeader(name) {
|
|
validateString(name, 'name');
|
|
return this[kOutHeaders] !== null &&
|
|
!!this[kOutHeaders][name.toLowerCase()];
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype.removeHeader = function removeHeader(name) {
|
|
validateString(name, 'name');
|
|
|
|
if (this._header) {
|
|
throw new ERR_HTTP_HEADERS_SENT('remove');
|
|
}
|
|
|
|
const key = name.toLowerCase();
|
|
|
|
switch (key) {
|
|
case 'connection':
|
|
this._removedConnection = true;
|
|
break;
|
|
case 'content-length':
|
|
this._removedContLen = true;
|
|
break;
|
|
case 'transfer-encoding':
|
|
this._removedTE = true;
|
|
break;
|
|
case 'date':
|
|
this.sendDate = false;
|
|
break;
|
|
}
|
|
|
|
if (this[kOutHeaders] !== null) {
|
|
delete this[kOutHeaders][key];
|
|
}
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype._implicitHeader = function _implicitHeader() {
|
|
this.emit('error', new ERR_METHOD_NOT_IMPLEMENTED('_implicitHeader()'));
|
|
};
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'headersSent', {
|
|
configurable: true,
|
|
enumerable: true,
|
|
get: function() { return !!this._header; }
|
|
});
|
|
|
|
Object.defineProperty(OutgoingMessage.prototype, 'writableEnded', {
|
|
get: function() { return this.finished; }
|
|
});
|
|
|
|
|
|
const crlf_buf = Buffer.from('\r\n');
|
|
OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
|
|
const ret = write_(this, chunk, encoding, callback, false);
|
|
if (!ret)
|
|
this[kNeedDrain] = true;
|
|
return ret;
|
|
};
|
|
|
|
function write_(msg, chunk, encoding, callback, fromEnd) {
|
|
if (msg.finished) {
|
|
const err = new ERR_STREAM_WRITE_AFTER_END();
|
|
const triggerAsyncId = msg.socket ? msg.socket[async_id_symbol] : undefined;
|
|
defaultTriggerAsyncIdScope(triggerAsyncId,
|
|
process.nextTick,
|
|
writeAfterEndNT,
|
|
msg,
|
|
err,
|
|
callback);
|
|
|
|
return true;
|
|
}
|
|
|
|
if (!msg._header) {
|
|
msg._implicitHeader();
|
|
}
|
|
|
|
if (!msg._hasBody) {
|
|
debug('This type of response MUST NOT have a body. ' +
|
|
'Ignoring write() calls.');
|
|
if (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[kIsCorked]) {
|
|
msg.socket.cork();
|
|
msg[kIsCorked] = true;
|
|
process.nextTick(connectionCorkNT, msg, msg.socket);
|
|
}
|
|
|
|
var len, ret;
|
|
if (msg.chunkedEncoding && chunk.length !== 0) {
|
|
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);
|
|
ret = msg._send(crlf_buf, null, callback);
|
|
} else {
|
|
ret = msg._send(chunk, encoding, callback);
|
|
}
|
|
|
|
debug('write ret = ' + ret);
|
|
return ret;
|
|
}
|
|
|
|
|
|
function writeAfterEndNT(msg, err, callback) {
|
|
msg.emit('error', err);
|
|
if (callback) callback(err);
|
|
}
|
|
|
|
|
|
function connectionCorkNT(msg, conn) {
|
|
msg[kIsCorked] = false;
|
|
conn.uncork();
|
|
}
|
|
|
|
|
|
OutgoingMessage.prototype.addTrailers = function addTrailers(headers) {
|
|
this._trailer = '';
|
|
const keys = Object.keys(headers);
|
|
const isArray = Array.isArray(headers);
|
|
var field, value;
|
|
for (var i = 0, l = keys.length; i < l; i++) {
|
|
var key = keys[i];
|
|
if (isArray) {
|
|
field = headers[key][0];
|
|
value = headers[key][1];
|
|
} else {
|
|
field = key;
|
|
value = headers[key];
|
|
}
|
|
if (typeof field !== 'string' || !field || !checkIsHttpToken(field)) {
|
|
throw new ERR_INVALID_HTTP_TOKEN('Trailer name', field);
|
|
}
|
|
if (checkInvalidHeaderChar(value)) {
|
|
debug('Trailer "%s" contains invalid characters', field);
|
|
throw new ERR_INVALID_CHAR('trailer content', field);
|
|
}
|
|
this._trailer += field + ': ' + value + CRLF;
|
|
}
|
|
};
|
|
|
|
function onFinish(outmsg) {
|
|
outmsg.emit('finish');
|
|
}
|
|
|
|
OutgoingMessage.prototype.end = function end(chunk, encoding, callback) {
|
|
if (typeof chunk === 'function') {
|
|
callback = chunk;
|
|
chunk = null;
|
|
} else if (typeof encoding === 'function') {
|
|
callback = encoding;
|
|
encoding = null;
|
|
}
|
|
|
|
if (this.finished) {
|
|
if (typeof callback === 'function') {
|
|
if (!this.writableFinished) {
|
|
this.on('finish', callback);
|
|
} else {
|
|
callback(new ERR_STREAM_ALREADY_FINISHED('end'));
|
|
}
|
|
}
|
|
return this;
|
|
}
|
|
|
|
var uncork;
|
|
if (chunk) {
|
|
if (typeof chunk !== 'string' && !(chunk instanceof Buffer)) {
|
|
throw new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer'], chunk);
|
|
}
|
|
if (!this._header) {
|
|
if (typeof chunk === 'string')
|
|
this._contentLength = Buffer.byteLength(chunk, encoding);
|
|
else
|
|
this._contentLength = chunk.length;
|
|
}
|
|
if (this.socket) {
|
|
this.socket.cork();
|
|
uncork = true;
|
|
}
|
|
write_(this, chunk, encoding, null, true);
|
|
} else if (!this._header) {
|
|
this._contentLength = 0;
|
|
this._implicitHeader();
|
|
}
|
|
|
|
if (typeof callback === 'function')
|
|
this.once('finish', callback);
|
|
|
|
const finish = onFinish.bind(undefined, this);
|
|
|
|
if (this._hasBody && this.chunkedEncoding) {
|
|
this._send('0\r\n' + this._trailer + '\r\n', 'latin1', finish);
|
|
} else {
|
|
// Force a flush, HACK.
|
|
this._send('', 'latin1', finish);
|
|
}
|
|
|
|
if (uncork)
|
|
this.socket.uncork();
|
|
|
|
this.finished = true;
|
|
|
|
// There is the first message on the outgoing queue, and we've sent
|
|
// everything to the socket.
|
|
debug('outgoing message end.');
|
|
if (this.outputData.length === 0 &&
|
|
this.socket &&
|
|
this.socket._httpMessage === this) {
|
|
this._finish();
|
|
}
|
|
|
|
return this;
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype._finish = function _finish() {
|
|
assert(this.socket);
|
|
this.emit('prefinish');
|
|
};
|
|
|
|
|
|
// This logic is probably a bit confusing. Let me explain a bit:
|
|
//
|
|
// In both HTTP servers and clients it is possible to queue up several
|
|
// outgoing messages. This is easiest to imagine in the case of a client.
|
|
// Take the following situation:
|
|
//
|
|
// req1 = client.request('GET', '/');
|
|
// req2 = client.request('POST', '/');
|
|
//
|
|
// When the user does
|
|
//
|
|
// req2.write('hello world\n');
|
|
//
|
|
// it's possible that the first request has not been completely flushed to
|
|
// the socket yet. Thus the outgoing messages need to be prepared to queue
|
|
// up data internally before sending it on further to the socket's queue.
|
|
//
|
|
// This function, outgoingFlush(), is called by both the Server and Client
|
|
// to attempt to flush any pending messages out to the socket.
|
|
OutgoingMessage.prototype._flush = function _flush() {
|
|
const socket = this.socket;
|
|
var ret;
|
|
|
|
if (socket && socket.writable) {
|
|
// There might be remaining data in this.output; write it out
|
|
ret = this._flushOutput(socket);
|
|
|
|
if (this.finished) {
|
|
// This is a queue to the server or client to bring in the next this.
|
|
this._finish();
|
|
} else if (ret && this[kNeedDrain]) {
|
|
this[kNeedDrain] = false;
|
|
this.emit('drain');
|
|
}
|
|
}
|
|
};
|
|
|
|
OutgoingMessage.prototype._flushOutput = function _flushOutput(socket) {
|
|
const outputLength = this.outputData.length;
|
|
if (outputLength <= 0)
|
|
return undefined;
|
|
|
|
const outputData = this.outputData;
|
|
socket.cork();
|
|
let ret;
|
|
for (var i = 0; i < outputLength; i++) {
|
|
const { data, encoding, callback } = outputData[i];
|
|
ret = socket.write(data, encoding, callback);
|
|
}
|
|
socket.uncork();
|
|
|
|
this.outputData = [];
|
|
this._onPendingData(-this.outputSize);
|
|
this.outputSize = 0;
|
|
|
|
return ret;
|
|
};
|
|
|
|
|
|
OutgoingMessage.prototype.flushHeaders = function flushHeaders() {
|
|
if (!this._header) {
|
|
this._implicitHeader();
|
|
}
|
|
|
|
// Force-flush the headers.
|
|
this._send('');
|
|
};
|
|
|
|
OutgoingMessage.prototype.flush = internalUtil.deprecate(function() {
|
|
this.flushHeaders();
|
|
}, 'OutgoingMessage.flush is deprecated. Use flushHeaders instead.', 'DEP0001');
|
|
|
|
OutgoingMessage.prototype.pipe = function pipe() {
|
|
// OutgoingMessage should be write-only. Piping from it is disabled.
|
|
this.emit('error', new ERR_STREAM_CANNOT_PIPE());
|
|
};
|
|
|
|
module.exports = {
|
|
OutgoingMessage
|
|
};
|