| 1276 | } |
| 1277 | |
| 1278 | const makeStreamPullEffect = <A, E>( |
| 1279 | get: AtomContext, |
| 1280 | pullSignal: Atom<number>, |
| 1281 | create: Stream.Stream<A, E, AtomRegistry> | ((get: AtomContext) => Stream.Stream<A, E, AtomRegistry>), |
| 1282 | options?: { |
| 1283 | readonly disableAccumulation?: boolean | undefined |
| 1284 | } |
| 1285 | ): Effect.Effect< |
| 1286 | { readonly done: boolean; readonly items: Arr.NonEmptyArray<A> }, |
| 1287 | E | Cause.NoSuchElementError, |
| 1288 | Scope.Scope | AtomRegistry |
| 1289 | > => |
| 1290 | Effect.flatMap( |
| 1291 | Stream.toPull(typeof create === "function" ? create(get) : create), |
| 1292 | (pullChunk) => { |
| 1293 | const fiber = Fiber.getCurrent()! |
| 1294 | const services = fiber.context as Context.Context<AtomRegistry | Scope.Scope> |
| 1295 | let acc: ReadonlyArray<A> = Arr.empty<A>() |
| 1296 | const pull: Effect.Effect< |
| 1297 | { |
| 1298 | done: boolean |
| 1299 | items: Arr.NonEmptyArray<A> |
| 1300 | }, |
| 1301 | Cause.NoSuchElementError | E, |
| 1302 | Registry.AtomRegistry |
| 1303 | > = Effect.matchCauseEffect(pullChunk, { |
| 1304 | onFailure(cause): Effect.Effect< |
| 1305 | { done: boolean; items: Arr.NonEmptyArray<A> }, |
| 1306 | Cause.NoSuchElementError | E |
| 1307 | > { |
| 1308 | if (Pull.isDoneCause(cause)) { |
| 1309 | if (!Arr.isReadonlyArrayNonEmpty(acc)) { |
| 1310 | return Effect.fail(new Cause.NoSuchElementError(`Atom.pull: no items`)) |
| 1311 | } |
| 1312 | return Effect.succeed({ done: true, items: acc as Arr.NonEmptyArray<A> }) |
| 1313 | } |
| 1314 | return Effect.failCause(cause as Cause.Cause<E>) |
| 1315 | }, |
| 1316 | onSuccess(chunk) { |
| 1317 | let items: Arr.NonEmptyArray<A> |
| 1318 | if (options?.disableAccumulation) { |
| 1319 | items = chunk as any |
| 1320 | } else { |
| 1321 | items = Arr.appendAll(acc, chunk) |
| 1322 | acc = items |
| 1323 | } |
| 1324 | return Effect.succeed({ done: false, items }) |
| 1325 | } |
| 1326 | }) |
| 1327 | |
| 1328 | const cancels = new Set<() => void>() |
| 1329 | get.addFinalizer(() => { |
| 1330 | for (const cancel of cancels) cancel() |
| 1331 | }) |
| 1332 | get.once(pullSignal) |
| 1333 | get.subscribe(pullSignal, () => { |
| 1334 | get.setSelf(AsyncResult.waitingFrom(get.self<PullResult<A, E>>())) |
| 1335 | let cancel: (() => void) | undefined |