Function unsafeStrategyCompletePollers

packages/effect/src/internal/pubsub.ts:1623–1650

(
  strategy: PubSubStrategy<A>,
  pubsub: AtomicPubSub<A>,
  subscribers: Subscribers<A>,
  subscription: Subscription<A>,
  pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>
)

/** @internal */
const unsafeStrategyCompletePollers = <A>(
  strategy: PubSubStrategy<A>,
  pubsub: AtomicPubSub<A>,
  subscribers: Subscribers<A>,
  subscription: Subscription<A>,
  pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>
): void => {
  let keepPolling = true
  while (keepPolling && !subscription.isEmpty()) {
    const poller = pipe(pollers, MutableQueue.poll(MutableQueue.EmptyMutableQueue))
    if (poller === MutableQueue.EmptyMutableQueue) {
      pipe(subscribers, removeSubscribers(subscription, pollers))
      if (MutableQueue.isEmpty(pollers)) {
        keepPolling = false
      } else {
        pipe(subscribers, addSubscribers(subscription, pollers))
      }
    } else {
      const pollResult = subscription.poll(MutableQueue.EmptyMutableQueue)
      if (pollResult === MutableQueue.EmptyMutableQueue) {
        unsafeOfferAll(pollers, pipe(unsafePollAllQueue(pollers), Chunk.prepend(poller)))
      } else {
        unsafeCompleteDeferred(poller, pollResult)
        strategy.unsafeOnPubSubEmptySpace(pubsub, subscribers)
      }
    }
  }
}

/** @internal */
const unsafeStrategyCompleteSubscribers = <A>(