Function takeAll

repos/effect/packages/effect/src/PubSub.ts:1208–1226

(self: Subscription<A>)

    } else if (!Arr.isArrayNonEmpty(as)) {
      return Effect.flatMap(pollForItem(self), (item) => loop([item]))
    }
    return Effect.succeed(as)
  })

const pollForItem = <A>(self: Subscription<A>) => {
  const deferred = Deferred.makeUnsafe<A>()
  let set = self.subscribers.get(self.subscription)
  if (!set) {
    set = new Set()
    self.subscribers.set(self.subscription, set)
  }
  set.add(self.pollers)
  MutableList.append(self.pollers, deferred)
  self.strategy.completePollersUnsafe(
    self.pubsub,
    self.subscribers,
    self.subscription,
    self.pollers
  )
  return Effect.onInterrupt(
    Deferred.await(deferred),
    () => {