(self: Subscription<A>)
| 1190 | * @since 4.0.0 |
| 1191 | */ |
| 1192 | export const takeAll = <A>(self: Subscription<A>): Effect.Effect<Arr.NonEmptyArray<A>> => |
| 1193 | Effect.suspend(function loop(value?: [A]): Effect.Effect<Arr.NonEmptyArray<A>> { |
| 1194 | if (self.shutdownFlag.current) { |
| 1195 | return Effect.interrupt |
| 1196 | } |
| 1197 | let as = self.pollers.length === 0 |
| 1198 | ? self.subscription.pollUpTo(Number.POSITIVE_INFINITY) |
| 1199 | : [] |
| 1200 | if (value) { |
| 1201 | as = value.concat(as) |
| 1202 | } |
| 1203 | self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers) |
| 1204 | if (self.replayWindow.remaining > 0) { |
| 1205 | return Effect.succeed(self.replayWindow.takeAll().concat(as) as Arr.NonEmptyArray<A>) |
| 1206 | } else if (!Arr.isArrayNonEmpty(as)) { |
| 1207 | return Effect.flatMap(pollForItem(self), (item) => loop([item])) |
| 1208 | } |
| 1209 | return Effect.succeed(as) |
| 1210 | }) |
| 1211 | |
| 1212 | const pollForItem = <A>(self: Subscription<A>) => { |
| 1213 | const deferred = Deferred.makeUnsafe<A>() |
nothing calls this directly
no test coverage detected