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
//
5
// Copyright Joyent, Inc. and other Node contributors.
6
//
7
// Permission is hereby granted, free of charge, to any person obtaining a
8
// copy of this software and associated documentation files (the
9
// "Software"), to deal in the Software without restriction, including
10
// without limitation the rights to use, copy, modify, merge, publish,
11
// distribute, sublicense, and/or sell copies of the Software, and to permit
12
// persons to whom the Software is furnished to do so, subject to the
13
// following conditions:
14
//
15
// The above copyright notice and this permission notice shall be included
16
// in all copies or substantial portions of the Software.
17
//
18
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
19
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
20
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
21
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
22
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
23
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
24
// USE OR OTHER DEALINGS IN THE SOFTWARE.
25
 
26
import {
27
  AbortError,
28
  ERR_INVALID_ARG_TYPE,
29
} from 'node-internal:internal_errors';
30
 
31
import {
32
  isNodeStream,
33
  isWebStream,
34
  kControllerErrorFunction,
35
} from 'node-internal:streams_util';
36
 
37
import { eos } from 'node-internal:streams_end_of_stream';
38
import type { Readable } from 'node-internal:streams_readable';
39
import type { Writable } from 'node-internal:streams_writable';
40
import type { Transform } from 'node-internal:streams_transform';
41
import { addAbortListener } from 'node-internal:events';
42
 
43
// This method is inlined here for readable-stream
44
// It also does not allow for signal to not exist on the stream
45
// https://github.com/nodejs/node/pull/36061#discussion_r533718029
46
function validateAbortSignal(
47
  signal: unknown,
48
  name: string
49
): asserts signal is AbortSignal {
50
  if (signal == null || typeof signal !== 'object' || !('aborted' in signal)) {
51
    throw new ERR_INVALID_ARG_TYPE(name, 'AbortSignal', signal);
52
  }
53
}
54
 
55
type StreamType =
56
  | Readable
57
  | Writable
58
  | Transform
59
  | ReadableStream
60
  | WritableStream;
61
 
62
export function addAbortSignal<T extends StreamType>(
63
  signal: unknown,
64
  stream: T
65
): T {
66
  validateAbortSignal(signal, 'signal');
67
  if (!isNodeStream(stream) && !isWebStream(stream)) {
68
    throw new ERR_INVALID_ARG_TYPE(
69
      'stream',
70
      ['ReadableStream', 'WritableStream', 'Stream'],
71
      stream
72
    );
73
  }
74
  return addAbortSignalNoValidate(signal, stream);
75
}
76
 
77
export function addAbortSignalNoValidate<T extends StreamType>(
78
  signal: AbortSignal | null | undefined,
79
  stream: T
80
): T {
81
  if (signal == null || typeof signal !== 'object' || !('aborted' in signal)) {
82
    return stream;
83
  }
84
  const onAbort = isNodeStream(stream)
85
    ? (): void => {
86
        stream.destroy(new AbortError(undefined, { cause: signal.reason }));
87
      }
88
    : (): void => {
89
        (
90
          stream as ReadableStream & {
91
            [kControllerErrorFunction]: (err: Error) => void;
92
          }
93
        )[kControllerErrorFunction](
94
          new AbortError(undefined, { cause: signal.reason })
95
        );
96
      };
97
  if (signal.aborted) {
98
    onAbort();
99
  } else {
100
    const disposable = addAbortListener(signal, onAbort);
101
    eos(
102
      stream as Readable | Writable | Transform,
103
      disposable[Symbol.dispose] as () => void
104
    );
105
  }
106
  return stream;
107
}