| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent 70e7541 commit f7adcd8
4 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -128,7 +128,7 @@ class BroadcastImpl { | |||
| 128 | 128 | // own internal AbortController that follows the external signal. | |
| 129 | 129 | // When no transforms, return rawConsumer directly (controller elided | |
| 130 | 130 | // per PULL-02 optimization -- no transforms means no signal recipient). | |
| 131 | - if (transforms.length > 0) { | ||
| 131 | + if (transforms.length > 0 || options?.signal) { | ||
| 132 | 132 | const pullArgs = [...transforms]; | |
| 133 | 133 | if (options?.signal) { | |
| 134 | 134 | ArrayPrototypePush(pullArgs, | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -97,7 +97,7 @@ class ShareImpl { | |||
| 97 | 97 | const { transforms, options } = parsePullArgs(args); | |
| 98 | 98 | const rawConsumer = this.#createRawConsumer(); | |
| 99 | 99 | ||
| 100 | - if (transforms.length > 0) { | ||
| 100 | + if (transforms.length > 0 || options?.signal) { | ||
| 101 | 101 | if (options) { | |
| 102 | 102 | return pullWithTransforms(rawConsumer, ...transforms, options); | |
| 103 | 103 | } | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -174,6 +174,19 @@ async function testPendingNextSettlesAfterReturn() { | |||
| 174 | 174 | assert.strictEqual(result.value, undefined); | |
| 175 | 175 | } | |
| 176 | 176 | ||
| 177 | + async function testPushAbortSignalRejectsPendingNext() { | ||
| 178 | + const ac = new AbortController(); | ||
| 179 | + const reason = new Error('push aborted'); | ||
| 180 | + const { broadcast: bc } = broadcast(); | ||
| 181 | + const iter = bc.push({ signal: ac.signal })[Symbol.asyncIterator](); | ||
| 182 | + | ||
| 183 | + const pendingNext = iter.next(); | ||
| 184 | + const rejected = assert.rejects(pendingNext, (error) => error === reason); | ||
| 185 | + ac.abort(reason); | ||
| 186 | + | ||
| 187 | + await rejected; | ||
| 188 | + } | ||
| 189 | + | ||
| 177 | 190 | // ============================================================================= | |
| 178 | 191 | // Writer fail detaches consumers | |
| 179 | 192 | // ============================================================================= | |
@@ -300,6 +313,7 @@ Promise.all([ | |||
| 300 | 313 | testCancelWithReason(), | |
| 301 | 314 | testCancelWithFalsyReason(), | |
| 302 | 315 | testPendingNextSettlesAfterReturn(), | |
| 316 | + testPushAbortSignalRejectsPendingNext(), | ||
| 303 | 317 | testFailDetachesConsumers(), | |
| 304 | 318 | testWriterFailIdempotent(), | |
| 305 | 319 | testLateJoinerSeesBufferedData(), | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -196,6 +196,25 @@ async function testShareAbortSignalWhileSourcePullPending() { | |||
| 196 | 196 | await Promise.all([rejected1, rejected2]); | |
| 197 | 197 | } | |
| 198 | 198 | ||
| 199 | + async function testSharePullAbortSignalRejectsPendingNext() { | ||
| 200 | + const ac = new AbortController(); | ||
| 201 | + const reason = new Error('pull aborted'); | ||
| 202 | + const shared = share( | ||
| 203 | + // eslint-disable-next-line require-yield | ||
| 204 | + (async function* never() { | ||
| 205 | + await new Promise(() => {}); | ||
| 206 | + })(), | ||
| 207 | + ); | ||
| 208 | + const iter = shared.pull({ signal: ac.signal })[Symbol.asyncIterator](); | ||
| 209 | + | ||
| 210 | + const pendingNext = iter.next(); | ||
| 211 | + const rejected = assert.rejects(pendingNext, (error) => error === reason); | ||
| 212 | + ac.abort(reason); | ||
| 213 | + | ||
| 214 | + await rejected; | ||
| 215 | + shared.cancel(); | ||
| 216 | + } | ||
| 217 | + | ||
| 199 | 218 | async function testShareAlreadyAborted() { | |
| 200 | 219 | const shared = share(from('data'), { signal: AbortSignal.abort() }); | |
| 201 | 220 | const consumer = shared.pull(); | |
@@ -340,6 +359,7 @@ Promise.all([ | |||
| 340 | 359 | testShareCancelWithReason(), | |
| 341 | 360 | testShareAbortSignal(), | |
| 342 | 361 | testShareAbortSignalWhileSourcePullPending(), | |
| 362 | + testSharePullAbortSignalRejectsPendingNext(), | ||
| 343 | 363 | testShareAlreadyAborted(), | |
| 344 | 364 | testShareSourceError(), | |
| 345 | 365 | testShareLateJoiningConsumer(), | |
| Back | FazBrowse Home | New Git URL |
0 commit comments