// @ts-check

'use strict'

/* global WebAssembly */

const assert = require('assert')

const net = require('net')

const http = require('http')

const { pipeline } = require('stream')

const util = require('./core/util')

const timers = require('./timers')

const Request = require('./core/request')

const DispatcherBase = require('./dispatcher-base')

const {

RequestContentLengthMismatchError,

ResponseContentLengthMismatchError,

InvalidArgumentError,

RequestAbortedError,

HeadersTimeoutError,

HeadersOverflowError,

SocketError,

InformationalError,

BodyTimeoutError,

HTTPParserError,

ResponseExceededMaxSizeError,

ClientDestroyedError

} = require('./core/errors')

const buildConnector = require('./core/connect')

const {

kUrl,

kReset,

kServerName,

kClient,

kBusy,

kParser,

kConnect,

kBlocking,

kResuming,

kRunning,

kPending,

kSize,

kWriting,

kQueue,

kConnected,

kConnecting,

kNeedDrain,

kNoRef,

kKeepAliveDefaultTimeout,

kHostHeader,

kPendingIdx,

kRunningIdx,

kError,

kPipelining,

kSocket,

kKeepAliveTimeoutValue,

kMaxHeadersSize,

kKeepAliveMaxTimeout,

kKeepAliveTimeoutThreshold,

kHeadersTimeout,

kBodyTimeout,

kStrictContentLength,

kConnector,

kMaxRedirections,

kMaxRequests,

kCounter,

kClose,

kDestroy,

kDispatch,

kInterceptors,

kLocalAddress,

kMaxResponseSize,

kHTTPConnVersion,

// HTTP2

kHost,

kHTTP2Session,

kHTTP2SessionState,

kHTTP2BuildRequest,

kHTTP2CopyHeaders,

kHTTP1BuildRequest

} = require('./core/symbols')

/** @type {import('http2')} */

let http2

try {

http2 = require('http2')

} catch {

// @ts-ignore

http2 = { constants: {} }

}

const {

constants: {

HTTP2_HEADER_AUTHORITY,

HTTP2_HEADER_METHOD,

HTTP2_HEADER_PATH,

HTTP2_HEADER_SCHEME,

HTTP2_HEADER_CONTENT_LENGTH,

HTTP2_HEADER_EXPECT,

HTTP2_HEADER_STATUS

}

} = http2

// Experimental

let h2ExperimentalWarned = false

const FastBuffer = Buffer[Symbol.species]

const kClosedResolve = Symbol('kClosedResolve')

const channels = {}

try {

const diagnosticsChannel = require('diagnostics_channel')

channels.sendHeaders = diagnosticsChannel.channel('undici:client:sendHeaders')

channels.beforeConnect = diagnosticsChannel.channel('undici:client:beforeConnect')

channels.connectError = diagnosticsChannel.channel('undici:client:connectError')

channels.connected = diagnosticsChannel.channel('undici:client:connected')

} catch {

channels.sendHeaders = { hasSubscribers: false }

channels.beforeConnect = { hasSubscribers: false }

channels.connectError = { hasSubscribers: false }

channels.connected = { hasSubscribers: false }

}

/**

* @type {import('../types/client').default}

*/

class Client extends DispatcherBase {

/**

*

* @param {string|URL} url

* @param {import('../types/client').Client.Options} options

*/

constructor (url, {

interceptors,

maxHeaderSize,

headersTimeout,

socketTimeout,

requestTimeout,

connectTimeout,

bodyTimeout,

idleTimeout,

keepAlive,

keepAliveTimeout,

maxKeepAliveTimeout,

keepAliveMaxTimeout,

keepAliveTimeoutThreshold,

socketPath,

pipelining,

tls,

strictContentLength,

maxCachedSessions,

maxRedirections,

connect,

maxRequestsPerClient,

localAddress,

maxResponseSize,

autoSelectFamily,

autoSelectFamilyAttemptTimeout,

// h2

allowH2,

maxConcurrentStreams

} = {}) {

super()

if (keepAlive !== undefined) {

throw new InvalidArgumentError('unsupported keepAlive, use pipelining=0 instead')

}

if (socketTimeout !== undefined) {

throw new InvalidArgumentError('unsupported socketTimeout, use headersTimeout & bodyTimeout instead')

}

if (requestTimeout !== undefined) {

throw new InvalidArgumentError('unsupported requestTimeout, use headersTimeout & bodyTimeout instead')

}

if (idleTimeout !== undefined) {

throw new InvalidArgumentError('unsupported idleTimeout, use keepAliveTimeout instead')

}

if (maxKeepAliveTimeout !== undefined) {

throw new InvalidArgumentError('unsupported maxKeepAliveTimeout, use keepAliveMaxTimeout instead')

}

if (maxHeaderSize != null && !Number.isFinite(maxHeaderSize)) {

throw new InvalidArgumentError('invalid maxHeaderSize')

}

if (socketPath != null && typeof socketPath !== 'string') {

throw new InvalidArgumentError('invalid socketPath')

}

if (connectTimeout != null && (!Number.isFinite(connectTimeout) || connectTimeout < 0)) {

throw new InvalidArgumentError('invalid connectTimeout')

}

if (keepAliveTimeout != null && (!Number.isFinite(keepAliveTimeout) || keepAliveTimeout <= 0)) {

throw new InvalidArgumentError('invalid keepAliveTimeout')

}

if (keepAliveMaxTimeout != null && (!Number.isFinite(keepAliveMaxTimeout) || keepAliveMaxTimeout <= 0)) {

throw new InvalidArgumentError('invalid keepAliveMaxTimeout')

}

if (keepAliveTimeoutThreshold != null && !Number.isFinite(keepAliveTimeoutThreshold)) {

throw new InvalidArgumentError('invalid keepAliveTimeoutThreshold')

}

if (headersTimeout != null && (!Number.isInteger(headersTimeout) || headersTimeout < 0)) {

throw new InvalidArgumentError('headersTimeout must be a positive integer or zero')

}

if (bodyTimeout != null && (!Number.isInteger(bodyTimeout) || bodyTimeout < 0)) {

throw new InvalidArgumentError('bodyTimeout must be a positive integer or zero')

}

if (connect != null && typeof connect !== 'function' && typeof connect !== 'object') {

throw new InvalidArgumentError('connect must be a function or an object')

}

if (maxRedirections != null && (!Number.isInteger(maxRedirections) || maxRedirections < 0)) {

throw new InvalidArgumentError('maxRedirections must be a positive number')

}

if (maxRequestsPerClient != null && (!Number.isInteger(maxRequestsPerClient) || maxRequestsPerClient < 0)) {

throw new InvalidArgumentError('maxRequestsPerClient must be a positive number')

}

if (localAddress != null && (typeof localAddress !== 'string' || net.isIP(localAddress) === 0)) {

throw new InvalidArgumentError('localAddress must be valid string IP address')

}

if (maxResponseSize != null && (!Number.isInteger(maxResponseSize) || maxResponseSize < -1)) {

throw new InvalidArgumentError('maxResponseSize must be a positive number')

}

if (

autoSelectFamilyAttemptTimeout != null &&

(!Number.isInteger(autoSelectFamilyAttemptTimeout) || autoSelectFamilyAttemptTimeout < -1)

) {

throw new InvalidArgumentError('autoSelectFamilyAttemptTimeout must be a positive number')

}

// h2

if (allowH2 != null && typeof allowH2 !== 'boolean') {

throw new InvalidArgumentError('allowH2 must be a valid boolean value')

}

if (maxConcurrentStreams != null && (typeof maxConcurrentStreams !== 'number' || maxConcurrentStreams < 1)) {

throw new InvalidArgumentError('maxConcurrentStreams must be a possitive integer, greater than 0')

}

if (typeof connect !== 'function') {

connect = buildConnector({

...tls,

maxCachedSessions,

allowH2,

socketPath,

timeout: connectTimeout,

...(util.nodeHasAutoSelectFamily && autoSelectFamily ? { autoSelectFamily, autoSelectFamilyAttemptTimeout } : undefined),

...connect

})

}

this[kInterceptors] = interceptors && interceptors.Client && Array.isArray(interceptors.Client)

? interceptors.Client

: [createRedirectInterceptor({ maxRedirections })]

this[kUrl] = util.parseOrigin(url)

this[kConnector] = connect

this[kSocket] = null

this[kPipelining] = pipelining != null ? pipelining : 1

this[kMaxHeadersSize] = maxHeaderSize || http.maxHeaderSize

this[kKeepAliveDefaultTimeout] = keepAliveTimeout == null ? 4e3 : keepAliveTimeout

this[kKeepAliveMaxTimeout] = keepAliveMaxTimeout == null ? 600e3 : keepAliveMaxTimeout

this[kKeepAliveTimeoutThreshold] = keepAliveTimeoutThreshold == null ? 1e3 : keepAliveTimeoutThreshold

this[kKeepAliveTimeoutValue] = this[kKeepAliveDefaultTimeout]

this[kServerName] = null

this[kLocalAddress] = localAddress != null ? localAddress : null

this[kResuming] = 0 // 0, idle, 1, scheduled, 2 resuming

this[kNeedDrain] = 0 // 0, idle, 1, scheduled, 2 resuming

this[kHostHeader] = `host: ${this[kUrl].hostname}${this[kUrl].port ? `:${this[kUrl].port}` : ''}\r\n`

this[kBodyTimeout] = bodyTimeout != null ? bodyTimeout : 300e3

this[kHeadersTimeout] = headersTimeout != null ? headersTimeout : 300e3

this[kStrictContentLength] = strictContentLength == null ? true : strictContentLength

this[kMaxRedirections] = maxRedirections

this[kMaxRequests] = maxRequestsPerClient

this[kClosedResolve] = null

this[kMaxResponseSize] = maxResponseSize > -1 ? maxResponseSize : -1

this[kHTTPConnVersion] = 'h1'

// HTTP/2

this[kHTTP2Session] = null

this[kHTTP2SessionState] = !allowH2

? null

: {

// streams: null, // Fixed queue of streams - For future support of `push`

openStreams: 0, // Keep track of them to decide wether or not unref the session

maxConcurrentStreams: maxConcurrentStreams != null ? maxConcurrentStreams : 100 // Max peerConcurrentStreams for a Node h2 server

}

this[kHost] = `${this[kUrl].hostname}${this[kUrl].port ? `:${this[kUrl].port}` : ''}`

// kQueue is built up of 3 sections separated by

// the kRunningIdx and kPendingIdx indices.

// | complete | running | pending |

// ^ kRunningIdx ^ kPendingIdx ^ kQueue.length

// kRunningIdx points to the first running element.

// kPendingIdx points to the first pending element.

// This implements a fast queue with an amortized

// time of O(1).

this[kQueue] = []

this[kRunningIdx] = 0

this[kPendingIdx] = 0

}

get pipelining () {

return this[kPipelining]

}

set pipelining (value) {

this[kPipelining] = value

resume(this, true)

}

get [kPending] () {

return this[kQueue].length - this[kPendingIdx]

}

get [kRunning] () {

return this[kPendingIdx] - this[kRunningIdx]

}

get [kSize] () {

return this[kQueue].length - this[kRunningIdx]

}

get [kConnected] () {

return !!this[kSocket] && !this[kConnecting] && !this[kSocket].destroyed

}

get [kBusy] () {

const socket = this[kSocket]

return (

(socket && (socket[kReset] || socket[kWriting] || socket[kBlocking])) ||

(this[kSize] >= (this[kPipelining] || 1)) ||

this[kPending] > 0

)

}

/* istanbul ignore: only used for test */

[kConnect] (cb) {

connect(this)

this.once('connect', cb)

}

[kDispatch] (opts, handler) {

const origin = opts.origin || this[kUrl].origin

const request = this[kHTTPConnVersion] === 'h2'

? Request[kHTTP2BuildRequest](origin, opts, handler)

: Request[kHTTP1BuildRequest](origin, opts, handler)

this[kQueue].push(request)

if (this[kResuming]) {

// Do nothing.

} else if (util.bodyLength(request.body) == null && util.isIterable(request.body)) {

// Wait a tick in case stream/iterator is ended in the same tick.

this[kResuming] = 1

process.nextTick(resume, this)

} else {

resume(this, true)

}

if (this[kResuming] && this[kNeedDrain] !== 2 && this[kBusy]) {

this[kNeedDrain] = 2

}

return this[kNeedDrain] < 2

}

async [kClose] () {

// TODO: for H2 we need to gracefully flush the remaining enqueued

// request and close each stream.

return new Promise((resolve) => {

if (!this[kSize]) {

resolve(null)

} else {

this[kClosedResolve] = resolve

}

})

}

async [kDestroy] (err) {

return new Promise((resolve) => {

const requests = this[kQueue].splice(this[kPendingIdx])

for (let i = 0; i < requests.length; i++) {

const request = requests[i]

errorRequest(this, request, err)

}

const callback = () => {

if (this[kClosedResolve]) {

// TODO (fix): Should we error here with ClientDestroyedError?

this[kClosedResolve]()

this[kClosedResolve] = null

}

resolve()

}

if (this[kHTTP2Session] != null) {

util.destroy(this[kHTTP2Session], err)

this[kHTTP2Session] = null

this[kHTTP2SessionState] = null

}

if (!this[kSocket]) {

queueMicrotask(callback)

} else {

util.destroy(this[kSocket].on('close', callback), err)

}

resume(this)

})

}

}

function onHttp2SessionError (err) {

assert(err.code !== 'ERR_TLS_CERT_ALTNAME_INVALID')

this[kSocket][kError] = err

onError(this[kClient], err)

}

function onHttp2FrameError (type, code, id) {

const err = new InformationalError(`HTTP/2: "frameError" received - type ${type}, code ${code}`)

if (id === 0) {

this[kSocket][kError] = err

onError(this[kClient], err)

}

}

function onHttp2SessionEnd () {

util.destroy(this, new SocketError('other side closed'))

util.destroy(this[kSocket], new SocketError('other side closed'))

}

function onHTTP2GoAway (code) {

const client = this[kClient]

const err = new InformationalError(`HTTP/2: "GOAWAY" frame received with code ${code}`)

client[kSocket] = null

client[kHTTP2Session] = null

if (client.destroyed) {

assert(this[kPending] === 0)

// Fail entire queue.

const requests = client[kQueue].splice(client[kRunningIdx])

for (let i = 0; i < requests.length; i++) {

const request = requests[i]

errorRequest(this, request, err)

}

} else if (client[kRunning] > 0) {

// Fail head of pipeline.

const request = client[kQueue][client[kRunningIdx]]

client[kQueue][client[kRunningIdx]++] = null

errorRequest(client, request, err)

}

client[kPendingIdx] = client[kRunningIdx]

assert(client[kRunning] === 0)

client.emit('disconnect',

client[kUrl],

[client],

err

)

resume(client)

}

const constants = require('./llhttp/constants')

const createRedirectInterceptor = require('./interceptor/redirectInterceptor')

const EMPTY_BUF = Buffer.alloc(0)

async function lazyllhttp () {

const llhttpWasmData = process.env.JEST_WORKER_ID ? require('./llhttp/llhttp-wasm.js') : undefined

let mod

try {

mod = await WebAssembly.compile(Buffer.from(require('./llhttp/llhttp_simd-wasm.js'), 'base64'))

} catch (e) {

/* istanbul ignore next */

// We could check if the error was caused by the simd option not

// being enabled, but the occurring of this other error

// * https://github.com/emscripten-core/emscripten/issues/11495

// got me to remove that check to avoid breaking Node 12.

mod = await WebAssembly.compile(Buffer.from(llhttpWasmData || require('./llhttp/llhttp-wasm.js'), 'base64'))

}

return await WebAssembly.instantiate(mod, {

env: {

/* eslint-disable camelcase */

wasm_on_url: (p, at, len) => {

/* istanbul ignore next */

return 0

},

wasm_on_status: (p, at, len) => {

assert.strictEqual(currentParser.ptr, p)

const start = at - currentBufferPtr + currentBufferRef.byteOffset

return currentParser.onStatus(new FastBuffer(currentBufferRef.buffer, start, len)) || 0

},

wasm_on_message_begin: (p) => {

assert.strictEqual(currentParser.ptr, p)

return currentParser.onMessageBegin() || 0

},

wasm_on_header_field: (p, at, len) => {

assert.strictEqual(currentParser.ptr, p)

const start = at - currentBufferPtr + currentBufferRef.byteOffset

return currentParser.onHeaderField(new FastBuffer(currentBufferRef.buffer, start, len)) || 0

},

wasm_on_header_value: (p, at, len) => {

assert.strictEqual(currentParser.ptr, p)

const start = at - currentBufferPtr + currentBufferRef.byteOffset

return currentParser.onHeaderValue(new FastBuffer(currentBufferRef.buffer, start, len)) || 0

},

wasm_on_headers_complete: (p, statusCode, upgrade, shouldKeepAlive) => {

assert.strictEqual(currentParser.ptr, p)

return currentParser.onHeadersComplete(statusCode, Boolean(upgrade), Boolean(shouldKeepAlive)) || 0

},

wasm_on_body: (p, at, len) => {

assert.strictEqual(currentParser.ptr, p)

const start = at - currentBufferPtr + currentBufferRef.byteOffset

return currentParser.onBody(new FastBuffer(currentBufferRef.buffer, start, len)) || 0

},

wasm_on_message_complete: (p) => {

assert.strictEqual(currentParser.ptr, p)

return currentParser.onMessageComplete() || 0

}

/* eslint-enable camelcase */

}

})

}

let llhttpInstance = null

let llhttpPromise = lazyllhttp()

llhttpPromise.catch()

let currentParser = null

let currentBufferRef = null

let currentBufferSize = 0

let currentBufferPtr = null

const TIMEOUT_HEADERS = 1

const TIMEOUT_BODY = 2

const TIMEOUT_IDLE = 3

class Parser {

constructor (client, socket, { exports }) {

assert(Number.isFinite(client[kMaxHeadersSize]) && client[kMaxHeadersSize] > 0)

this.llhttp = exports

this.ptr = this.llhttp.llhttp_alloc(constants.TYPE.RESPONSE)

this.client = client

this.socket = socket

this.timeout = null

this.timeoutValue = null

this.timeoutType = null

this.statusCode = null

this.statusText = ''

this.upgrade = false

this.headers = []

this.headersSize = 0

this.headersMaxSize = client[kMaxHeadersSize]

this.shouldKeepAlive = false

this.paused = false

this.resume = this.resume.bind(this)

this.bytesRead = 0

this.keepAlive = ''

this.contentLength = ''

this.connection = ''

this.maxResponseSize = client[kMaxResponseSize]

}

setTimeout (value, type) {

this.timeoutType = type

if (value !== this.timeoutValue) {

timers.clearTimeout(this.timeout)

if (value) {

this.timeout = timers.setTimeout(onParserTimeout, value, this)

// istanbul ignore else: only for jest

if (this.timeout.unref) {

this.timeout.unref()

}

} else {

this.timeout = null

}

this.timeoutValue = value

} else if (this.timeout) {

// istanbul ignore else: only for jest

if (this.timeout.refresh) {

this.timeout.refresh()

}

}

}

resume () {

if (this.socket.destroyed || !this.paused) {

return

}

assert(this.ptr != null)

assert(currentParser == null)

this.llhttp.llhttp_resume(this.ptr)

assert(this.timeoutType === TIMEOUT_BODY)

if (this.timeout) {

// istanbul ignore else: only for jest

if (this.timeout.refresh) {

this.timeout.refresh()

}

}

this.paused = false

this.execute(this.socket.read() || EMPTY_BUF) // Flush parser.

this.readMore()

}

readMore () {

while (!this.paused && this.ptr) {

const chunk = this.socket.read()

if (chunk === null) {

break

}

this.execute(chunk)

}

}

execute (data) {

assert(this.ptr != null)

assert(currentParser == null)

assert(!this.paused)

const { socket, llhttp } = this

if (data.length > currentBufferSize) {

if (currentBufferPtr) {

llhttp.free(currentBufferPtr)

}

currentBufferSize = Math.ceil(data.length / 4096) * 4096

currentBufferPtr = llhttp.malloc(currentBufferSize)

}

new Uint8Array(llhttp.memory.buffer, currentBufferPtr, currentBufferSize).set(data)

// Call `execute` on the wasm parser.

// We pass the `llhttp_parser` pointer address, the pointer address of buffer view data,

// and finally the length of bytes to parse.

// The return value is an error code or `constants.ERROR.OK`.

try {

let ret

try {

currentBufferRef = data

currentParser = this

ret = llhttp.llhttp_execute(this.ptr, currentBufferPtr, data.length)

/* eslint-disable-next-line no-useless-catch */

} catch (err) {

/* istanbul ignore next: difficult to make a test case for */

throw err

} finally {

currentParser = null

currentBufferRef = null

}

const offset = llhttp.llhttp_get_error_pos(this.ptr) - currentBufferPtr

if (ret === constants.ERROR.PAUSED_UPGRADE) {

this.onUpgrade(data.slice(offset))

} else if (ret === constants.ERROR.PAUSED) {

this.paused = true

socket.unshift(data.slice(offset))

} else if (ret !== constants.ERROR.OK) {

const ptr = llhttp.llhttp_get_error_reason(this.ptr)

let message = ''

/* istanbul ignore else: difficult to make a test case for */

if (ptr) {

const len = new Uint8Array(llhttp.memory.buffer, ptr).indexOf(0)

message =

'Response does not match the HTTP/1.1 protocol (' +

Buffer.from(llhttp.memory.buffer, ptr, len).toString() +

')'

}

throw new HTTPParserError(message, constants.ERROR[ret], data.slice(offset))

}

} catch (err) {

util.destroy(socket, err)

}

}

destroy () {

assert(this.ptr != null)

assert(currentParser == null)

this.llhttp.llhttp_free(this.ptr)

this.ptr = null

timers.clearTimeout(this.timeout)

this.timeout = null

this.timeoutValue = null

this.timeoutType = null

this.paused = false

}

onStatus (buf) {

this.statusText = buf.toString()

}

onMessageBegin () {

const { socket, client } = this

/* istanbul ignore next: difficult to make a test case for */

if (socket.destroyed) {

return -1

}

const request = client[kQueue][client[kRunningIdx]]

if (!request) {

return -1

}

}

onHeaderField (buf) {

const len = this.headers.length

if ((len & 1) === 0) {

this.headers.push(buf)

} else {

this.headers[len - 1] = Buffer.concat([this.headers[len - 1], buf])

}

this.trackHeader(buf.length)

}

onHeaderValue (buf) {

let len = this.headers.length

if ((len & 1) === 1) {

this.headers.push(buf)

len += 1

} else {

this.headers[len - 1] = Buffer.concat([this.headers[len - 1], buf])

}

const key = this.headers[len - 2]

if (key.length === 10 && key.toString().toLowerCase() === 'keep-alive') {

this.keepAlive += buf.toString()

} else if (key.length === 10 && key.toString().toLowerCase() === 'connection') {

this.connection += buf.toString()

} else if (key.length === 14 && key.toString().toLowerCase() === 'content-length') {

this.contentLength += buf.toString()

}

this.trackHeader(buf.length)

}

trackHeader (len) {

this.headersSize += len

if (this.headersSize >= this.headersMaxSize) {

util.destroy(this.socket, new HeadersOverflowError())

}

}

onUpgrade (head) {

const { upgrade, client, socket, headers, statusCode } = this

assert(upgrade)

const request = client[kQueue][client[kRunningIdx]]

assert(request)

assert(!socket.destroyed)

assert(socket === client[kSocket])

assert(!this.paused)

assert(request.upgrade || request.method === 'CONNECT')

this.statusCode = null

this.statusText = ''

this.shouldKeepAlive = null

assert(this.headers.length % 2 === 0)

this.headers = []

this.headersSize = 0

socket.unshift(head)

socket[kParser].destroy()

socket[kParser] = null

socket[kClient] = null

socket[kError] = null

socket

.removeListener('error', onSocketError)

.removeListener('readable', onSocketReadable)

.removeListener('end', onSocketEnd)

.removeListener('close', onSocketClose)

client[kSocket] = null

client[kQueue][client[kRunningIdx]++] = null

client.emit('disconnect', client[kUrl], [client], new InformationalError('upgrade'))

try {

request.onUpgrade(statusCode, headers, socket)

} catch (err) {

util.destroy(socket, err)

}

resume(client)

}

onHeadersComplete (statusCode, upgrade, shouldKeepAlive) {

const { client, socket, headers, statusText } = this

/* istanbul ignore next: difficult to make a test case for */

if (socket.destroyed) {

return -1

}

const request = client[kQueue][client[kRunningIdx]]

/* istanbul ignore next: difficult to make a test case for */

if (!request) {

return -1

}

assert(!this.upgrade)

assert(this.statusCode < 200)

if (statusCode === 100) {

util.destroy(socket, new SocketError('bad response', util.getSocketInfo(socket)))

return -1

}

/* this can only happen if server is misbehaving */

if (upgrade && !request.upgrade) {

util.destroy(socket, new SocketError('bad upgrade', util.getSocketInfo(socket)))

return -1

}

assert.strictEqual(this.timeoutType, TIMEOUT_HEADERS)

this.statusCode = statusCode

this.shouldKeepAlive = (

shouldKeepAlive ||

// Override llhttp value which does not allow keepAlive for HEAD.

(request.method === 'HEAD' && !socket[kReset] && this.connection.toLowerCase() === 'keep-alive')

)

if (this.statusCode >= 200) {

const bodyTimeout = request.bodyTimeout != null

? request.bodyTimeout

: client[kBodyTimeout]

this.setTimeout(bodyTimeout, TIMEOUT_BODY)

} else if (this.timeout) {

// istanbul ignore else: only for jest

if (this.timeout.refresh) {

this.timeout.refresh()

}

}

if (request.method === 'CONNECT') {

assert(client[kRunning] === 1)

this.upgrade = true

return 2

}

if (upgrade) {

assert(client[kRunning] === 1)

this.upgrade = true

return 2

}

assert(this.headers.length % 2 === 0)

this.headers = []

this.headersSize = 0

if (this.shouldKeepAlive && client[kPipelining]) {

const keepAliveTimeout = this.keepAlive ? util.parseKeepAliveTimeout(this.keepAlive) : null

if (keepAliveTimeout != null) {

const timeout = Math.min(

keepAliveTimeout - client[kKeepAliveTimeoutThreshold],

client[kKeepAliveMaxTimeout]

)

if (timeout <= 0) {

socket[kReset] = true

} else {

client[kKeepAliveTimeoutValue] = timeout

}

} else {

client[kKeepAliveTimeoutValue] = client[kKeepAliveDefaultTimeout]

}

} else {

// Stop more requests from being dispatched.

socket[kReset] = true

}

let pause

try {

pause = request.onHeaders(statusCode, headers, this.resume, statusText) === false

} catch (err) {

util.destroy(socket, err)

return -1

}

if (request.method === 'HEAD') {

return 1

}

if (statusCode < 200) {

return 1

}

if (socket[kBlocking]) {

socket[kBlocking] = false

resume(client)

}

return pause ? constants.ERROR.PAUSED : 0

}

onBody (buf) {

const { client, socket, statusCode, maxResponseSize } = this

if (socket.destroyed) {

return -1

}

const request = client[kQueue][client[kRunningIdx]]

assert(request)

assert.strictEqual(this.timeoutType, TIMEOUT_BODY)

if (this.timeout) {

// istanbul ignore else: only for jest

if (this.timeout.refresh) {

this.timeout.refresh()

}

}

assert(statusCode >= 200)

if (maxResponseSize > -1 && this.bytesRead + buf.length > maxResponseSize) {

util.destroy(socket, new ResponseExceededMaxSizeError())

return -1

}

this.bytesRead += buf.length

try {

if (request.onData(buf) === false) {

return constants.ERROR.PAUSED

}

} catch (err) {

util.destroy(socket, err)

return -1

}

}

onMessageComplete () {

const { client, socket, statusCode, upgrade, headers, contentLength, bytesRead, shouldKeepAlive } = this

if (socket.destroyed && (!statusCode || shouldKeepAlive)) {

return -1

}

if (upgrade) {

return

}

const request = client[kQueue][client[kRunningIdx]]

assert(request)

assert(statusCode >= 100)

this.statusCode = null

this.statusText = ''

this.bytesRead = 0

this.contentLength = ''