1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167 |
x7
x7
x7
x30
x62
x47
x55
x55
x30
x22
x22
x22
x22
x22
x22
x66
x88
x37
x22
x22
x15
x15
x15
x15
x15
x15
x45
x15
x15
x15
x112
x28
x31
x28
x28
x28
x30
x30
x30
x30
x48
x47
x15
x18
x18
x18
x15 |
|
// Copyright 2018-2025 the Deno authors. MIT license.
// This module is browser compatible.
// TODO(iuioiua): Remove `ignore` directives from following snippets
/**
* Make a {@linkcode Promise} abortable with the given signal.
*
* @throws {DOMException} If the signal is already aborted and `signal.reason`
* is undefined. Otherwise, throws `signal.reason`.
* @typeParam T The type of the provided and returned promise.
* @param p The promise to make abortable.
* @param signal The signal to abort the promise with.
* @returns A promise that can be aborted.
*
* @example Error-handling a timeout
* ```ts ignore
* import { abortable, delay } from "@std/async";
* import { assertRejects, assertEquals } from "@std/assert";
*
* const promise = delay(1_000);
*
* // Rejects with `DOMException` after 100 ms
* await assertRejects(
* () => abortable(promise, AbortSignal.timeout(100)),
* DOMException,
* "Signal timed out."
* );
* ```
*
* @example Error-handling an abort
* ```ts ignore
* import { abortable, delay } from "@std/async";
* import { assertRejects, assertEquals } from "@std/assert";
*
* const promise = delay(1_000);
* const controller = new AbortController();
* controller.abort(new Error("This is my reason"));
*
* // Rejects with `DOMException` immediately
* await assertRejects(
* () => abortable(promise, controller.signal),
* Error,
* "This is my reason"
* );
* ```
*/
export function abortable<T>(p: Promise<T>, signal: AbortSignal): Promise<T>;
/**
* Make an {@linkcode AsyncIterable} abortable with the given signal.
*
* @throws {DOMException} If the signal is already aborted and `signal.reason`
* is undefined. Otherwise, throws `signal.reason`.
* @typeParam T The type of the provided and returned async iterable.
* @param p The async iterable to make abortable.
* @param signal The signal to abort the promise with.
* @returns An async iterable that can be aborted.
*
* @example Error-handling a timeout
* ```ts
* import { abortable, delay } from "@std/async";
* import { assertRejects, assertEquals } from "@std/assert";
*
* const asyncIter = async function* () {
* yield "Hello";
* await delay(1_000);
* yield "World";
* };
*
* const items: string[] = [];
* // Below throws `DOMException` after 100 ms and items become `["Hello"]`
* await assertRejects(
* async () => {
* for await (const item of abortable(asyncIter(), AbortSignal.timeout(100))) {
* items.push(item);
* }
* },
* DOMException,
* "Signal timed out."
* );
* assertEquals(items, ["Hello"]);
* ```
*
* @example Error-handling an abort
* ```ts
* import { abortable, delay } from "@std/async";
* import { assertRejects, assertEquals } from "@std/assert";
*
* const asyncIter = async function* () {
* yield "Hello";
* await delay(1_000);
* yield "World";
* };
* const controller = new AbortController();
* controller.abort(new Error("This is my reason"));
*
* const items: string[] = [];
* // Below throws `DOMException` immediately
* await assertRejects(
* async () => {
* for await (const item of abortable(asyncIter(), controller.signal)) {
* items.push(item);
* }
* },
* Error,
* "This is my reason"
* );
* assertEquals(items, []);
* ```
*/
export function abortable<T>(
p: AsyncIterable<T>,
signal: AbortSignal,
): AsyncGenerator<T>;
export function abortable<T>(
p: Promise<T> | AsyncIterable<T>,
signal: AbortSignal,
): Promise<T> | AsyncIterable<T> {
if (p instanceof Promise) {
return abortablePromise(p, signal);
} else {
return abortableAsyncIterable(p, signal);
}
}
function abortablePromise<T>(
p: Promise<T>,
signal: AbortSignal,
): Promise<T> {
const { promise, reject } = Promise.withResolvers<never>();
const abort = () => reject(signal.reason);
if (signal.aborted) abort();
signal.addEventListener("abort", abort, { once: true });
return Promise.race([promise, p]).finally(() => {
signal.removeEventListener("abort", abort);
});
}
async function* abortableAsyncIterable<T>(
p: AsyncIterable<T>,
signal: AbortSignal,
): AsyncGenerator<T> {
signal.throwIfAborted();
const { promise, reject } = Promise.withResolvers<never>();
const abort = () => reject(signal.reason);
signal.addEventListener("abort", abort, { once: true });
const it = p[Symbol.asyncIterator]();
try {
while (true) {
const race = Promise.race([promise, it.next()]);
race.catch(() => {
signal.removeEventListener("abort", abort);
});
const { done, value } = await race;
if (done) {
signal.removeEventListener("abort", abort);
const result = await it.return?.(value);
return result?.value;
}
yield value;
}
} catch (e) {
await it.return?.();
throw e;
}
}
|