| 1329 | }) |
| 1330 | |
| 1331 | const testOuterFailure = ( |
| 1332 | combinator: "flatMap" | "switchMap", |
| 1333 | inner: "never" | "slow" |
| 1334 | ) => |
| 1335 | Effect.gen(function*() { |
| 1336 | const started = yield* Latch.make(false) |
| 1337 | const failing = yield* Deferred.make<void>() |
| 1338 | const finalized = yield* Ref.make(0) |
| 1339 | const outer = Stream.concat( |
| 1340 | Stream.make(1), |
| 1341 | Stream.fromEffect( |
| 1342 | started.await.pipe( |
| 1343 | Effect.andThen(Deferred.succeed(failing, void 0)), |
| 1344 | Effect.andThen(Effect.fail("boom")) |
| 1345 | ) |
| 1346 | ) |
| 1347 | ) |
| 1348 | const makeInner = () => { |
| 1349 | started.openUnsafe() |
| 1350 | return (inner === "never" ? Stream.never : Stream.fromEffect(Effect.sleep(Duration.hours(1)))).pipe( |
| 1351 | Stream.ensuring(Ref.update(finalized, (n) => n + 1)) |
| 1352 | ) |
| 1353 | } |
| 1354 | const stream = combinator === "flatMap" |
| 1355 | ? Stream.flatMap(outer, makeInner, { concurrency: 2 }) |
| 1356 | : Stream.switchMap(outer, makeInner) |
| 1357 | const fiber = yield* stream.pipe( |
| 1358 | Stream.runDrain, |
| 1359 | Effect.forkChild |
| 1360 | ) |
| 1361 | yield* Deferred.await(failing) |
| 1362 | |
| 1363 | const result = yield* Fiber.await(fiber) |
| 1364 | assert.deepStrictEqual(result, Exit.fail("boom")) |
| 1365 | assert.strictEqual(yield* Ref.get(finalized), 1) |
| 1366 | }) |
| 1367 | |
| 1368 | describe("flatMap", () => { |
| 1369 | it.effect("interrupts all inner streams when the outer fails at the concurrency limit", () => |