Skip to content
rachel/workerd
File

Blob: src/node/internal/internal_http_outgoing.ts

typescript1241 lines
1// Copyright (c) 2017-2022 Cloudflare, Inc.
2// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3//     https://opensource.org/licenses/Apache-2.0
4// Copyright Joyent and Node contributors. All rights reserved. MIT license.
5 
6// We have deprecations because @types/node defines this.finished as deprecated.
7/* eslint-disable @typescript-eslint/no-deprecated */
8 
9import { validateString } from 'node-internal:validators';
10import { Writable } from 'node-internal:streams_writable';
11import { getDefaultHighWaterMark } from 'node-internal:streams_state';
12import type { DataWrittenEvent } from 'node-internal:internal_http_server';
13import {
14  ERR_HTTP_HEADERS_SENT,
15  ERR_INVALID_ARG_TYPE,
16  ERR_STREAM_CANNOT_PIPE,
17  ERR_STREAM_DESTROYED,
18  ERR_METHOD_NOT_IMPLEMENTED,
19  ERR_STREAM_WRITE_AFTER_END,
20  ERR_HTTP_CONTENT_LENGTH_MISMATCH,
21  ERR_HTTP_BODY_NOT_ALLOWED,
22  ERR_STREAM_NULL_VALUES,
23  ERR_STREAM_ALREADY_FINISHED,
24  ERR_INVALID_ARG_VALUE,
25} from 'node-internal:internal_errors';
26import { isUint8Array } from 'node-internal:internal_types';
27import {
28  validateHeaderName,
29  validateHeaderValue,
30  chunkExpression as RE_TE_CHUNKED,
31  utcDate,
32} from 'node-internal:internal_http';
33import { IncomingMessage } from 'node-internal:internal_http_incoming';
34import { EventEmitter } from 'node-internal:events';
35import type {
36  OutgoingMessage as _OutgoingMessage,
37  OutgoingHttpHeaders,
38  ServerResponse,
39  OutgoingHttpHeader,
40} from 'node:http';
41 
42type WriteCallback = (err?: Error) => void;
43export type OutputData = {
44  data: string | Buffer | Uint8Array | null;
45  encoding?: BufferEncoding | null | undefined;
46  callback?: WriteCallback | null | undefined;
47};
48export type WrittenDataBufferEntry = OutputData & {
49  length: number;
50  written: boolean;
51};
52export type HeadersSentEvent = {
53  statusCode: number;
54  statusMessage: string;
55  headers: Headers;
56};
57 
58export const kUniqueHeaders = Symbol('kUniqueHeaders');
59export const kHighWaterMark = Symbol('kHighWaterMark');
60export const kNeedDrain = Symbol('kNeedDrain');
61export const kOutHeaders = Symbol('kOutHeaders');
62export const kErrored = Symbol('kErrored');
63const kCorked = Symbol('corked');
64const kChunkedBuffer = Symbol('kChunkedBuffer');
65const kChunkedLength = Symbol('kChunkedLength');
66const kBytesWritten = Symbol('kBytesWritten');
67const kRejectNonStandardBodyWrites = Symbol('kRejectNonStandardBodyWrites');
68 
69const RE_CONN_CLOSE = /(?:^|\W)close(?:$|\W)/i;
70 
71type HeaderState = {
72  connection: boolean;
73  contLen: boolean;
74  te: boolean;
75  date: boolean;
76  expect: boolean;
77  trailer: boolean;
78  header: string;
79};
80 
81export function parseUniqueHeadersOption(
82  headers?: (string | string[])[]
83): Set<string> | null {
84  if (!Array.isArray(headers)) {
85    return null;
86  }
87 
88  const unique = new Set<string>();
89  for (const header of headers) {
90    if (Array.isArray(header)) {
91      for (const h of header) {
92        unique.add(h.toLowerCase());
93      }
94    } else {
95      unique.add(header.toLowerCase());
96    }
97  }
98  return unique;
99}
100 
101// Most of the code in this class is derived from Michael Hart's project
102// Ref: https://github.com/mhart/fetch-to-node/blob/main/src/fetch-to-node/http-outgoing.ts
103class MessageBuffer {
104  #corked = 0;
105  #index = 0;
106  #onWrite: (data: DataWrittenEvent[]) => void;
107  #bufferedWrites: { index: number; entry: WrittenDataBufferEntry }[] = [];
108  #highWaterMark: number;
109 
110  constructor(
111    onWrite: (data: DataWrittenEvent[]) => void,
112    options: { highWaterMark: number }
113  ) {
114    this.#onWrite = onWrite;
115    this.#highWaterMark = options.highWaterMark;
116  }
117 
118  write(
119    data: WrittenDataBufferEntry['data'],
120    encoding: WrittenDataBufferEntry['encoding'],
121    callback: WrittenDataBufferEntry['callback'],
122    onDrain?: (dataLength: number) => void
123  ): boolean {
124    const entry: WrittenDataBufferEntry = {
125      data,
126      length: data?.length ?? 0,
127      encoding,
128      callback,
129      written: true,
130    };
131 
132    const index = this.#index++;
133 
134    if (this.#corked === 0) {
135      this.#onWrite([{ index, entry }]);
136      queueMicrotask(() => {
137        onDrain?.(entry.length);
138        callback?.();
139      });
140    } else {
141      // Buffer the write when corked
142      this.#bufferedWrites.push({ index, entry });
143      queueMicrotask(() => {
144        callback?.();
145      });
146    }
147 
148    return true;
149  }
150 
151  cork(): void {
152    this.#corked++;
153  }
154 
155  uncork(): void {
156    this.#corked--;
157    this._flush();
158  }
159 
160  _flush(): void {
161    // If fully uncorked, flush all buffered writes
162    if (this.#corked <= 0) {
163      this.#onWrite(this.#bufferedWrites.splice(0));
164    }
165  }
166 
167  get writableLength(): number {
168    return this.#bufferedWrites.reduce((total, { entry }) => {
169      return total + (entry.length || 0);
170    }, 0);
171  }
172 
173  get writableHighWaterMark(): number {
174    return this.#highWaterMark;
175  }
176 
177  get writableCorked(): number {
178    return this.#corked;
179  }
180}
181 
182export type OutgoingMessageOptions = {
183  highWaterMark?: number | undefined;
184  rejectNonStandardBodyWrites?: boolean | undefined;
185};
186 
187// Most of the code in this class is derived from Michael Hart's project
188// Ref: https://github.com/mhart/fetch-to-node/blob/main/src/fetch-to-node/http-outgoing.ts
189export class OutgoingMessage extends Writable implements _OutgoingMessage {
190  [kOutHeaders]: Record<string, [string, string | string[]]> | null = null;
191  [kErrored]: Error | null = null;
192  [kCorked] = 0;
193  [kChunkedBuffer]: OutputData[] = [];
194  [kChunkedLength]: number = 0;
195  [kNeedDrain] = false;
196  [kRejectNonStandardBodyWrites]: boolean;
197  [kHighWaterMark]: number;
198  [kBytesWritten] = 0;
199 
200  // @ts-expect-error TS2416 IncomingMessage is not feature complete yet.
201  readonly req?: IncomingMessage | undefined;
202  #buffer: MessageBuffer | undefined | null;
203 
204  // Queue that holds all currently pending data, until the response will be
205  // assigned to the socket (until it will its turn in the HTTP pipeline).
206  outputData: OutputData[] = [];
207 
208  // `outputSize` is an approximate measure of how much data is queued on this
209  // response. `_onPendingData` will be invoked to update similar global
210  // per-connection counter. That counter will be used to pause/unpause the
211  // TCP socket and HTTP Parser and thus handle the backpressure.
212  outputSize: number = 0;
213 
214  // `writtenHeaderBytes` is the number of bytes the header has taken.
215  // Since Node.js writes both the headers and body into the same outgoing
216  // stream, it helps to keep track of this so that we can skip that many bytes
217  // from the beginning of the stream when providing the outgoing stream.
218  writtenHeaderBytes = 0;
219 
220  strictContentLength = false;
221  chunkedEncoding = false;
222  sendDate = false;
223  shouldKeepAlive = true;
224  override writable = true;
225  finished = false;
226  override destroyed = false;
227  useChunkedEncodingByDefault = true;
228  maxRequestsOnConnectionReached = false;
229 
230  // These are attributes provided by the Node.js implementation.
231  override _closed = false;
232  _headerSent = false;
233  _onPendingData: (delta: number) => void = () => {};
234  _header: string | null = null;
235  _contentLength: number | null = null;
236  _hasBody = true;
237  _removedContLen = false;
238  _removedConnection = false;
239  _removedTE = false;
240  _last = false;
241  _defaultKeepAlive = true;
242  _maxRequestsPerSocket: number | undefined;
243  _keepAliveTimeout = 0;
244 
245  constructor(req?: IncomingMessage, options?: OutgoingMessageOptions) {
246    super();
247    this.req = req;
248    this[kHighWaterMark] = options?.highWaterMark ?? getDefaultHighWaterMark();
249    this[kRejectNonStandardBodyWrites] =
250      options?.rejectNonStandardBodyWrites ?? false;
251    this.#buffer = new MessageBuffer(this.#onDataWritten.bind(this), {
252      highWaterMark: this[kHighWaterMark],
253    });
254 
255    this.once('end', () => {
256      // We need to emit close in a queueMicrotask because
257      // this is the only way we can ensure that the close event is emitted after destroy.
258      queueMicrotask(() => {
259        this._closed = true;
260        this.emit('close');
261      });
262    });
263  }
264 
265  #onDataWritten(data: DataWrittenEvent[]): void {
266    this.emit('_dataWritten', data);
267  }
268 
269  override cork(): void {
270    this[kCorked]++;
271    this.#buffer?.cork();
272  }
273 
274  override uncork(): void {
275    this[kCorked]--;
276    this.#buffer?.uncork();
277 
278    if (this[kCorked] || this[kChunkedBuffer].length === 0) {
279      return;
280    }
281 
282    for (const { data, encoding, callback } of this[kChunkedBuffer]) {
283      this._send(data ?? '', encoding, callback);
284    }
285 
286    this[kChunkedBuffer].length = 0;
287    this[kChunkedLength] = 0;
288  }
289 
290  _storeHeader(
291    firstLine: string,
292    headers: OutgoingHttpHeaders | OutgoingHttpHeader[] | null
293  ): void {
294    // firstLine in the case of request is: 'GET /index.html HTTP/1.1\r\n'
295    // in the case of response it is: 'HTTP/1.1 200 OK\r\n'
296    const state: HeaderState = {
297      connection: false,
298      contLen: false,
299      te: false,
300      date: false,
301      expect: false,
302      trailer: false,
303      header: firstLine,
304    };
305 
306    if (headers != null) {
307      if (headers === this[kOutHeaders]) {
308        for (const key in headers) {
309          const entry = headers[key] as [string, string];
310          processHeader(this, state, entry[0], entry[1], false);
311        }
312      } else if (Array.isArray(headers)) {
313        if (headers.length && Array.isArray(headers[0])) {
314          for (let i = 0; i < headers.length; i++) {
315            const entry = headers[i] as unknown as [string, string];
316            processHeader(this, state, entry[0], entry[1], true);
317          }
318        } else {
319          if (headers.length % 2 !== 0) {
320            throw new ERR_INVALID_ARG_VALUE('headers', headers);
321          }
322 
323          for (let n = 0; n < headers.length; n += 2) {
324            processHeader(
325              this,
326              state,
327              headers[n] as string,
328              headers[n + 1] as string,
329              true
330            );
331          }
332        }
333      } else {
334        for (const key in headers) {
335          // eslint-disable-next-line no-prototype-builtins
336          if (headers.hasOwnProperty(key)) {
337            const _headers = headers;
338            processHeader(
339              this,
340              state,
341              key,
342              _headers[key] as OutgoingHttpHeader,
343              true
344            );
345          }
346        }
347      }
348    }
349 
350    let { header } = state;
351 
352    // Date header
353    if (this.sendDate && !state.date) {
354      header += 'Date: ' + utcDate() + '\r\n';
355    }
356 
357    // Force the connection to close when the response is a 204 No Content or
358    // a 304 Not Modified and the user has set a "Transfer-Encoding: chunked"
359    // header.
360    //
361    // RFC 2616 mandates that 204 and 304 responses MUST NOT have a body but
362    // node.js used to send out a zero chunk anyway to accommodate clients
363    // that don't have special handling for those responses.
364    //
365    // It was pointed out that this might confuse reverse proxies to the point
366    // of creating security liabilities, so suppress the zero chunk and force
367    // the connection to close.
368    if (
369      this.chunkedEncoding &&
370      ((this as unknown as ServerResponse).statusCode === 204 ||
371        (this as unknown as ServerResponse).statusCode === 304)
372    ) {
373      this.chunkedEncoding = false;
374      this.shouldKeepAlive = false;
375    }
376 
377    // keep-alive logic
378    if (this._removedConnection) {
379      // shouldKeepAlive is generally true for HTTP/1.1. In that common case,
380      // even if the connection header isn't sent, we still persist by default.
381      this._last = !this.shouldKeepAlive;
382    } else if (!state.connection) {
383      const shouldSendKeepAlive =
384        this.shouldKeepAlive &&
385        (state.contLen ||
386          this.useChunkedEncodingByDefault ||
387          (this as unknown as { agent: unknown }).agent);
388      if (shouldSendKeepAlive && this.maxRequestsOnConnectionReached) {
389        header += 'Connection: close\r\n';
390      } else if (shouldSendKeepAlive) {
391        header += 'Connection: keep-alive\r\n';
392        if (this._keepAliveTimeout && this._defaultKeepAlive) {
393          const timeoutSeconds = Math.floor(this._keepAliveTimeout / 1000);
394          let max = '';
395          if (
396            this._maxRequestsPerSocket != null &&
397            ~~this._maxRequestsPerSocket > 0
398          ) {
399            max = `, max=${this._maxRequestsPerSocket}`;
400          }
401          header += `Keep-Alive: timeout=${timeoutSeconds}${max}\r\n`;
402        }
403      } else {
404        this._last = true;
405        header += 'Connection: close\r\n';
406      }
407    }
408 
409    if (!state.contLen && !state.te) {
410      if (!this._hasBody) {
411        // Make sure we don't end the 0\r\n\r\n at the end of the message.
412        this.chunkedEncoding = false;
413      } else if (!this.useChunkedEncodingByDefault) {
414        this._last = true;
415      } else if (
416        !state.trailer &&
417        !this._removedContLen &&
418        typeof this._contentLength === 'number'
419      ) {
420        // eslint-disable-next-line @typescript-eslint/restrict-plus-operands
421        header += 'Content-Length: ' + this._contentLength + '\r\n';
422      } else if (!this._removedTE) {
423        header += 'Transfer-Encoding: chunked\r\n';
424        this.chunkedEncoding = true;
425      } else {
426        // We can't keep alive in this case, because with no header info the body
427        // is defined as all data until the connection is closed.
428        this._last = true;
429      }
430    }
431 
432    this._header = header + '\r\n';
433    this._headerSent = false;
434 
435    // Wait until the first body chunk, or close(), is sent to flush,
436    // UNLESS we're sending Expect: 100-continue.
437    if (state.expect) this._send('');
438  }
439 
440  _finish(): void {
441    this.emit('prefinish');
442  }
443 
444  _flushOutput(buffer: MessageBuffer): boolean | undefined {
445    const outputData = this.outputData;
446    if (outputData.length === 0) {
447      return undefined;
448    }
449 
450    buffer.cork();
451    for (const { data, encoding, callback } of outputData) {
452      buffer.write(data, encoding, callback);
453    }
454    buffer.uncork();
455 
456    this.outputData = [];
457    this._onPendingData(-this.outputSize);
458    this.outputSize = 0;
459 
460    return true;
461  }
462 
463  _flush(): void {
464    if (this.#buffer != null) {
465      const ret = this._flushOutput(this.#buffer);
466 
467      if (this.finished) {
468        this._finish();
469      } else if (ret && this[kNeedDrain]) {
470        this[kNeedDrain] = false;
471        this.emit('drain');
472      }
473    }
474  }
475 
476  // @ts-expect-error TS2611 Required for accessor
477  get writableLength(): number {
478    // If using buffer with headers, include buffer's writable length
479    if (this.#buffer != null && (this._header !== null || this._headerSent)) {
480      return (
481        this.outputSize + this.#buffer.writableLength + this[kChunkedLength]
482      );
483    }
484    return this.outputSize + this[kChunkedLength];
485  }
486 
487  // @ts-expect-error TS2611 Required for accessor
488  get writableCorked(): number {
489    return this[kCorked];
490  }
491 
492  // @ts-expect-error TS2611 Required for accessor
493  get writableNeedDrain(): boolean {
494    return !this.destroyed && !this.finished && this[kNeedDrain];
495  }
496 
497  setHeader(name: string, value: number | string | string[]): this {
498    if (this._header) {
499      throw new ERR_HTTP_HEADERS_SENT('set');
500    }
501    validateHeaderName(name);
502    validateHeaderValue(name, value);
503 
504    let headers = this[kOutHeaders];
505    if (headers === null) {
506      this[kOutHeaders] = headers = {};
507    }
508 
509    headers[name.toLowerCase()] = [name, value];
510    return this;
511  }
512  setHeaders(
513    headers: Headers | Map<string, string | number | readonly string[]>
514  ): this {
515    if (this.headersSent) {
516      throw new ERR_HTTP_HEADERS_SENT('set');
517    }
518 
519    if (
520      Array.isArray(headers) ||
521      typeof headers !== 'object' ||
522      !('keys' in headers) ||
523      !('get' in headers) ||
524      typeof headers.keys !== 'function' ||
525      typeof headers.get !== 'function'
526    ) {
527      throw new ERR_INVALID_ARG_TYPE('headers', ['Headers', 'Map'], headers);
528    }
529 
530    // Headers object joins multiple cookies with a comma when using
531    // the getter to retrieve the value,
532    // unless iterating over the headers directly.
533    // We also cannot safely split by comma.
534    // To avoid setHeader overwriting the previous value we push
535    // set-cookie values in array and set them all at once.
536    const cookies: string[] = [];
537 
538    for (const { 0: key, 1: value } of headers) {
539      if (key === 'set-cookie') {
540        if (Array.isArray(value)) {
541          cookies.push(...(value as string[]));
542        } else {
543          cookies.push(value as string);
544        }
545        continue;
546      }
547      this.setHeader(key, value as string | string[]);
548    }
549    if (cookies.length) {
550      this.setHeader('set-cookie', cookies);
551    }
552 
553    return this;
554  }
555 
556  appendHeader(
557    name: string,
558    value: number | string | ReadonlyArray<string> | OutgoingHttpHeader
559  ): this {
560    if (this._header) {
561      throw new ERR_HTTP_HEADERS_SENT('append');
562    }
563    validateHeaderName(name);
564    validateHeaderValue(name, value);
565 
566    const field = name.toLowerCase();
567    const headers = this[kOutHeaders];
568    if (headers === null || !headers[field]) {
569      return this.setHeader(name, value);
570    }
571 
572    // Prepare the field for appending, if required
573    if (!Array.isArray(headers[field][1])) {
574      headers[field][1] = [headers[field][1]];
575    }
576 
577    const existingValues = headers[field][1];
578    if (Array.isArray(value)) {
579      for (let i = 0, length = value.length; i < length; i++) {
580        existingValues.push(value[i] as string);
581      }
582    } else {
583      existingValues.push(value);
584    }
585 
586    return this;
587  }
588 
589  getHeader(name: string): number | string | string[] | undefined {
590    validateString(name, 'name');
591 
592    const headers = this[kOutHeaders];
593    if (headers === null) {
594      return;
595    }
596 
597    const entry = headers[name.toLowerCase()];
598    return entry?.[1] as string;
599  }
600 
601  hasHeader(name: unknown): boolean {
602    validateString(name, 'name');
603    return Boolean(this[kOutHeaders]?.[name.toLowerCase()]);
604  }
605 
606  removeHeader(name: string): void {
607    validateString(name, 'name');
608 
609    if (this._header) {
610      throw new ERR_HTTP_HEADERS_SENT('remove');
611    }
612 
613    const key = name.toLowerCase();
614 
615    switch (key) {
616      case 'connection':
617        this._removedConnection = true;
618        break;
619      case 'content-length':
620        this._removedContLen = true;
621        break;
622      case 'transfer-encoding':
623        this._removedTE = true;
624        break;
625      case 'date':
626        this.sendDate = false;
627        break;
628    }
629 
630    if (this[kOutHeaders] !== null) {
631      // eslint-disable-next-line @typescript-eslint/no-dynamic-delete
632      delete this[kOutHeaders][key];
633    }
634  }
635 
636  // Returns an array of the names of the current outgoing headers.
637  getHeaderNames(): string[] {
638    return this[kOutHeaders] !== null ? Object.keys(this[kOutHeaders]) : [];
639  }
640 
641  // Returns an array of the names of the current outgoing raw headers.
642  getRawHeaderNames(): string[] {
643    const headersMap = this[kOutHeaders];
644    if (headersMap === null) return [];
645 
646    const values = Object.values(headersMap);
647    const headers = Array.from<string>({ length: values.length });
648    // Retain for(;;) loop for performance reasons
649    // Refs: https://github.com/nodejs/node/pull/30958
650    for (let i = 0, l = values.length; i < l; i++) {
651      headers[i] = (values[i] as [string, string])[0];
652    }
653 
654    return headers;
655  }
656 
657  flushHeaders(): void {
658    if (!this._header) {
659      this._implicitHeader();
660    }
661 
662    // Force-flush the headers.
663    this._send('');
664  }
665 
666  getHeaders(): OutgoingHttpHeaders {
667    const headers = this[kOutHeaders];
668    const ret: Record<string, string> = {};
669    if (headers) {
670      const keys = Object.keys(headers);
671      // Retain for(;;) loop for performance reasons
672      // Refs: https://github.com/nodejs/node/pull/30958
673      for (let i = 0; i < keys.length; ++i) {
674        const key = keys[i] as keyof typeof headers;
675        const val = (headers[key] as [string, string])[1];
676        ret[key] = val;
677      }
678    }
679    return ret;
680  }
681 
682  get headersSent(): boolean {
683    return !!this._header;
684  }
685 
686  override pipe<T extends NodeJS.WritableStream>(destination: T): T {
687    this.emit('error', new ERR_STREAM_CANNOT_PIPE());
688    return destination;
689  }
690 
691  [EventEmitter.captureRejectionSymbol](error: Error): void {
692    this.destroy(error);
693  }
694 
695  _implicitHeader(): void {
696    throw new ERR_METHOD_NOT_IMPLEMENTED('_implicitHeader()');
697  }
698 
699  _renderHeaders(): Record<string, string> {
700    if (this._header) {
701      throw new ERR_HTTP_HEADERS_SENT('render');
702    }
703 
704    const headersMap = this[kOutHeaders];
705    const headers: Record<string, string> = {};
706 
707    if (headersMap !== null) {
708      const keys = Object.keys(headersMap);
709      // Retain for(;;) loop for performance reasons
710      // Refs: https://github.com/nodejs/node/pull/30958
711      for (let i = 0; i < keys.length; i++) {
712        const key = keys[i] as keyof typeof headersMap;
713        headers[(headersMap[key] as [string, string])[0]] = (
714          headersMap[key] as [string, string]
715        )[1];
716      }
717    }
718    return headers;
719  }
720 
721  _send(
722    data: string | Uint8Array,
723    encoding?: BufferEncoding | WriteCallback | null,
724    callback?: WriteCallback | null,
725    byteLength?: number
726  ): boolean {
727    // This is a shameful hack to get the headers and first body chunk onto
728    // the same packet. Future versions of Node are going to take care of
729    // this at a lower level and in a more general way.
730    if (!this._headerSent && this._header !== null) {
731      const header = this._header;
732      if (
733        typeof data === 'string' &&
734        (encoding === 'utf8' || encoding === 'latin1' || !encoding)
735      ) {
736        data = header + data;
737      } else {
738        this.outputData.unshift({
739          data: header,
740          encoding: 'latin1',
741          callback: null,
742        });
743        this.outputSize += header.length;
744        this._onPendingData(header.length);
745      }
746 
747      this._headerSent = true;
748      this.writtenHeaderBytes = header.length;
749 
750      // Difference from Node.js: Parse response headers to emit _headersSent event.
751      // This deviates from Node.js behavior but is required for ServerResponse compatibility.
752      //
753      // The same OutgoingMessage class is used for both:
754      // - Client requests: header starts with request line "POST /path HTTP/1.1"
755      // - Server responses: header starts with status line "HTTP/1.1 200 OK"
756      //
757      // We only parse and emit events for server responses (status lines that match the HTTP response format).
758      // Client requests are ignored to avoid parsing errors when request lines don't match response format.
759      const [statusLine, ...headerLines] = header.split('\r\n') as [
760        string,
761        ...string[],
762      ];
763 
764      const STATUS_LINE_REGEXP =
765        /^HTTP\/\d+\.\d+ (?<statusCode>\d+) (?<statusMessage>.*)$/;
766      const statusLineResult = STATUS_LINE_REGEXP.exec(statusLine);
767 
768      if (statusLineResult != null) {
769        const { statusCode: statusCodeText, statusMessage } =
770          statusLineResult.groups ?? {};
771        const headers = new Headers();
772 
773        for (const headerLine of headerLines) {
774          if (headerLine !== '') {
775            const pos = headerLine.indexOf(': ');
776            headers.append(headerLine.slice(0, pos), headerLine.slice(pos + 2));
777          }
778        }
779        this.emit('_headersSent', {
780          statusCode: Number(statusCodeText as string),
781          statusMessage,
782          headers,
783        } as HeadersSentEvent);
784      }
785    }
786    return this._writeRaw(data, encoding, callback, byteLength);
787  }
788 
789  override write(
790    chunk: string | Buffer | Uint8Array,
791    encoding?: BufferEncoding | WriteCallback | null,
792    callback?: WriteCallback
793  ): boolean {
794    if (typeof encoding === 'function') {
795      callback = encoding;
796      encoding = null;
797    }
798 
799    const ret = this.#write(chunk, encoding, callback, false);
800    if (!ret) {
801      this[kNeedDrain] = true;
802    }
803    return ret;
804  }
805 
806  override end(
807    chunk?: string | Buffer | Uint8Array | WriteCallback | null,
808    encoding?: BufferEncoding | WriteCallback | null,
809    callback?: WriteCallback
810  ): this {
811    if (typeof chunk === 'function') {
812      callback = chunk;
813      chunk = null;
814      encoding = null;
815    } else if (typeof encoding === 'function') {
816      callback = encoding;
817      encoding = null;
818    }
819 
820    if (chunk) {
821      if (this.finished) {
822        this.#onError(
823          new ERR_STREAM_WRITE_AFTER_END(),
824          typeof callback !== 'function' ? (): void => {} : callback
825        );
826        return this;
827      }
828 
829      // Difference from Node.js -
830      // In Node.js, if a socket exists, we would also call socket.cork() at this point.
831      // For our implementation we do the same for the "written data buffer"
832      this.#buffer?.cork();
833      this.#write(chunk, encoding, null, true);
834    } else if (this.finished) {
835      if (typeof callback === 'function') {
836        if (!this.writableFinished) {
837          this.on('finish', callback);
838        } else {
839          callback(new ERR_STREAM_ALREADY_FINISHED('end'));
840        }
841      }
842      return this;
843    } else if (!this._header) {
844      // Difference from Node.js -
845      // In Node.js, if a socket exists, we would also call socket.cork() at this point.
846      // For our implementation we do the same for the "written data buffer"
847      this.#buffer?.cork();
848      this._contentLength = 0;
849      this._implicitHeader();
850    }
851 
852    if (typeof callback === 'function') this.once('finish', callback);
853 
854    if (
855      this.#checkStrictContentLength() &&
856      this[kBytesWritten] !== this._contentLength
857    ) {
858      throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(
859        this[kBytesWritten],
860        this._contentLength ?? 0
861      );
862    }
863 
864    const finish = onFinish.bind(undefined, this);
865 
866    if (this._hasBody && this.chunkedEncoding) {
867      // Difference from Node.js -
868      // Chunked transfer encoding doesn't need to use the low-level protocol
869      // (with each chunk preceded by its length)
870      // So here we just send an empty chunk. Trailers are not supported
871 
872      // this._send("0\r\n" + this._trailer + "\r\n", "latin1", finish);
873      this._send('', 'latin1', finish);
874    } else if (!this._headerSent || this.writableLength || chunk) {
875      this._send('', 'latin1', finish);
876    } else {
877      queueMicrotask(finish);
878    }
879 
880    // Difference from Node.js -
881    // In Node.js, if a socket exists, we would also call socket.uncork() at this point.
882    // For our implementation we do the same for the "written data buffer"
883    this.#buffer?.uncork();
884    this[kCorked] = 1;
885    this.uncork();
886 
887    this.finished = true;
888    this._writableState.finished = true;
889    this._writableState.corked = 1;
890 
891    // Difference from Node.js -
892    // In Node.js, if a socket exists, and there is no pending output data,
893    // we would also call this._finish() at this point.
894    // For our implementation we do the same for the "written data buffer"
895    if (this.outputData.length === 0 && this.#buffer != null) {
896      this._finish();
897    }
898 
899    return this;
900  }
901 
902  _writeRaw(
903    data: string | Uint8Array,
904    encoding?: BufferEncoding | WriteCallback | null,
905    callback?: WriteCallback | null,
906    _size?: number
907  ): boolean {
908    if (this.destroyed) {
909      return false;
910    }
911 
912    // Difference from Node.js -
913    // In Node.js, we would check for an underlying socket, and if that socket
914    // exists and is already destroyed, simply return false.
915 
916    if (typeof encoding === 'function') {
917      callback = encoding;
918      encoding = null;
919    }
920 
921    // Difference from Node.js -
922    // In Node.js, we would check for an underlying socket, and if that socket
923    // exists and is currently writable, it would flush any pending data to the socket and then
924    // write the current chunk's data directly into the socket. Afterwards, it would return with the
925    // value returned from socket.write().
926    if (this.#buffer != null && (this._header !== null || this._headerSent)) {
927      if (this.outputData.length) {
928        this._flushOutput(this.#buffer);
929      }
930 
931      this.#buffer.write(data, encoding, callback);
932 
933      // Always return true for corked writes (imitating Node.js behavior)
934      if (this.#buffer.writableCorked > 0) {
935        return true;
936      }
937 
938      // For uncorked writes, check if we need to signal backpressure
939      // based on the buffer's high water mark
940      return this.#buffer.writableLength < this.#buffer.writableHighWaterMark;
941    }
942 
943    this.outputData.push({ data, encoding, callback });
944    this.outputSize += data.length;
945    this._onPendingData(data.length);
946    return this.outputSize < this[kHighWaterMark];
947  }
948 
949  override destroy(err?: unknown, _cb?: (err?: unknown) => void): this {
950    if (this.destroyed) {
951      return this;
952    }
953    if (err != null) {
954      this.emit('error', err);
955    }
956    this.destroyed = true;
957    this[kErrored] = err as Error;
958 
959    return this;
960  }
961 
962  // @ts-expect-error TS2611 Property accessor.
963  get errored(): Error | null {
964    return this[kErrored];
965  }
966 
967  // @ts-expect-error TS2611 Property accessor.
968  get closed(): boolean {
969    return this._closed;
970  }
971 
972  // @ts-expect-error TS2611 Property accessor.
973  get writableEnded(): boolean {
974    return this.finished;
975  }
976 
977  // @ts-expect-error TS2611 Property accessor.
978  get writableHighWaterMark(): number {
979    return this.#buffer?.writableHighWaterMark ?? this[kHighWaterMark];
980  }
981 
982  // @ts-expect-error TS2611 Property accessor.
983  get writableObjectMode(): boolean {
984    return false;
985  }
986 
987  #checkStrictContentLength(): boolean {
988    return (
989      this.strictContentLength &&
990      this._contentLength != null &&
991      this._hasBody &&
992      !this._removedContLen &&
993      !this.chunkedEncoding &&
994      !this.hasHeader('transfer-encoding')
995    );
996  }
997 
998  #onError(err: Error, callback: WriteCallback): void {
999    if (this.destroyed) {
1000      return;
1001    }
1002 
1003    queueMicrotask(() => {
1004      emitErrorNt(this, err, callback);
1005    });
1006  }
1007 
1008  #write(
1009    this: OutgoingMessage,
1010    chunk: string | Buffer | Uint8Array | null,
1011    encoding: BufferEncoding | undefined | null,
1012    callback: WriteCallback | undefined | null,
1013    fromEnd: boolean
1014  ): boolean {
1015    if (typeof callback !== 'function') {
1016      callback = (): void => {};
1017    }
1018 
1019    if (chunk === null) {
1020      throw new ERR_STREAM_NULL_VALUES();
1021    } else if (typeof chunk !== 'string' && !isUint8Array(chunk)) {
1022      throw new ERR_INVALID_ARG_TYPE(
1023        'chunk',
1024        ['string', 'Buffer', 'Uint8Array'],
1025        chunk
1026      );
1027    }
1028 
1029    let err: Error | undefined = undefined;
1030    if (this.finished) {
1031      err = new ERR_STREAM_WRITE_AFTER_END();
1032    } else if (this.destroyed) {
1033      err = new ERR_STREAM_DESTROYED('write');
1034    }
1035 
1036    if (err) {
1037      if (!this.destroyed) {
1038        this.#onError(err, callback);
1039      } else {
1040        queueMicrotask(() => {
1041          callback(err);
1042        });
1043      }
1044      return false;
1045    }
1046 
1047    let len: number | undefined = undefined;
1048 
1049    if (this.strictContentLength) {
1050      len ??=
1051        typeof chunk === 'string'
1052          ? Buffer.byteLength(chunk, encoding ?? undefined)
1053          : chunk.byteLength;
1054 
1055      if (
1056        this.#checkStrictContentLength() &&
1057        (fromEnd
1058          ? this[kBytesWritten] + len !== this._contentLength
1059          : this[kBytesWritten] + len > (this._contentLength ?? 0))
1060      ) {
1061        throw new ERR_HTTP_CONTENT_LENGTH_MISMATCH(
1062          len + this[kBytesWritten],
1063          this._contentLength ?? 0
1064        );
1065      }
1066 
1067      this[kBytesWritten] += len;
1068    }
1069 
1070    if (!this._header) {
1071      if (fromEnd) {
1072        len ??=
1073          typeof chunk === 'string'
1074            ? Buffer.byteLength(chunk, encoding ?? undefined)
1075            : chunk.byteLength;
1076        this._contentLength = len;
1077      }
1078      this._implicitHeader();
1079    }
1080 
1081    if (!this._hasBody) {
1082      if (this[kRejectNonStandardBodyWrites]) {
1083        throw new ERR_HTTP_BODY_NOT_ALLOWED();
1084      } else {
1085        queueMicrotask(callback);
1086        return true;
1087      }
1088    }
1089 
1090    if (!fromEnd && this.#buffer != null && !this.#buffer.writableCorked) {
1091      this.#buffer.cork();
1092      queueMicrotask(() => {
1093        connectionCorkNT(this.#buffer as MessageBuffer);
1094      });
1095    }
1096 
1097    let ret;
1098    if (this.chunkedEncoding && chunk.length !== 0) {
1099      len ??=
1100        typeof chunk === 'string'
1101          ? Buffer.byteLength(chunk, encoding ?? undefined)
1102          : chunk.byteLength;
1103      if (this[kCorked] && this._headerSent) {
1104        this[kChunkedBuffer].push({ data: chunk, encoding, callback });
1105        this[kChunkedLength] += len;
1106        ret = this[kChunkedLength] < this[kHighWaterMark];
1107      } else {
1108        ret = this._send(chunk, encoding, callback, len);
1109      }
1110    } else {
1111      ret = this._send(chunk, encoding, callback, len);
1112    }
1113 
1114    return ret;
1115  }
1116}
1117 
1118function emitErrorNt(
1119  msg: OutgoingMessage,
1120  err: Error,
1121  callback: WriteCallback
1122): void {
1123  callback(err);
1124  if (typeof msg.emit === 'function' && !msg.destroyed) {
1125    msg.emit('error', err);
1126  }
1127}
1128 
1129function onFinish(outmsg: OutgoingMessage): void {
1130  outmsg.emit('finish');
1131}
1132 
1133function connectionCorkNT(buffer: MessageBuffer): void {
1134  buffer.uncork();
1135}
1136 
1137// isCookieField performs a case-insensitive comparison of a provided string
1138// against the word "cookie." As of V8 6.6 this is faster than handrolling or
1139// using a case-insensitive RegExp.
1140function isCookieField(s: string): boolean {
1141  return s.length === 6 && s.toLowerCase() === 'cookie';
1142}
1143 
1144function isContentDispositionField(s: string): boolean {
1145  return s.length === 19 && s.toLowerCase() === 'content-disposition';
1146}
1147 
1148function processHeader(
1149  self: OutgoingMessage,
1150  state: HeaderState,
1151  key: string,
1152  value: OutgoingHttpHeader,
1153  validate: boolean
1154): void {
1155  if (validate) {
1156    validateHeaderName(key);
1157  }
1158 
1159  // If key is content-disposition and there is content-length
1160  // encode the value in latin1
1161  // https://www.rfc-editor.org/rfc/rfc6266#section-4.3
1162  // Refs: https://github.com/nodejs/node/pull/46528
1163  if (isContentDispositionField(key) && self._contentLength) {
1164    // The value could be an array here
1165    if (Array.isArray(value)) {
1166      for (let i = 0; i < value.length; i++) {
1167        value[i] = String(Buffer.from(String(value[i]), 'latin1'));
1168      }
1169    } else {
1170      value = String(Buffer.from(String(value), 'latin1'));
1171    }
1172  }
1173 
1174  if (Array.isArray(value)) {
1175    if (
1176      (value.length < 2 || !isCookieField(key)) &&
1177      (!(kUniqueHeaders in self) ||
1178        !(self[kUniqueHeaders] as Set<string>).has(key.toLowerCase()))
1179    ) {
1180      // Retain for(;;) loop for performance reasons
1181      // Refs: https://github.com/nodejs/node/pull/30958
1182      for (let i = 0; i < value.length; i++) {
1183        storeHeader(self, state, key, value[i] as string, validate);
1184      }
1185      return;
1186    }
1187    value = value.join('; ');
1188  }
1189  storeHeader(self, state, key, String(value), validate);
1190}
1191 
1192function storeHeader(
1193  self: OutgoingMessage,
1194  state: HeaderState,
1195  key: string,
1196  value: string,
1197  validate: boolean
1198): void {
1199  if (validate) {
1200    validateHeaderValue(key, value);
1201  }
1202  state.header += key + ': ' + value + '\r\n';
1203  matchHeader(self, state, key, value);
1204}
1205 
1206function matchHeader(
1207  self: OutgoingMessage,
1208  state: HeaderState,
1209  field: string,
1210  value: string
1211): void {
1212  if (field.length < 4 || field.length > 17) return;
1213  field = field.toLowerCase();
1214  switch (field) {
1215    case 'connection':
1216      state.connection = true;
1217      self._removedConnection = false;
1218      if (RE_CONN_CLOSE.exec(value) !== null) self._last = true;
1219      else self.shouldKeepAlive = true;
1220      break;
1221    case 'transfer-encoding':
1222      state.te = true;
1223      self._removedTE = false;
1224      if (RE_TE_CHUNKED.exec(value) !== null) self.chunkedEncoding = true;
1225      break;
1226    case 'content-length':
1227      state.contLen = true;
1228      self._contentLength = +value;
1229      self._removedContLen = false;
1230      break;
1231    case 'date':
1232    case 'expect':
1233    case 'trailer':
1234      state[field] = true;
1235      break;
1236    case 'keep-alive':
1237      self._defaultKeepAlive = false;
1238      break;
1239  }
1240}