Class BackPressureStrategy

packages/effect/src/internal/pubsub.ts:1399–1509

* A strategy that applies back pressure to publishers when the `PubSub` is at * capacity. This guarantees that all subscribers will receive all messages * published to the `PubSub` while they are subscribed. However, it creates the * risk that a slow subscriber will slow down the rate at which me

 * @internal
 */
class BackPressureStrategy<in out A> implements PubSubStrategy<A> {
  publishers: MutableQueue.MutableQueue<
    readonly [A, Deferred.Deferred<boolean>, boolean]
  > = MutableQueue.unbounded()

  get shutdown(): Effect.Effect<void> {
    return core.flatMap(core.fiberId, (fiberId) =>
      core.flatMap(
        core.sync(() => unsafePollAllQueue(this.publishers)),
        (publishers) =>
          fiberRuntime.forEachConcurrentDiscard(
            publishers,
            ([_, deferred, last]) =>
              last ?
                pipe(core.deferredInterruptWith(deferred, fiberId), core.asVoid) :
                core.void,
            false,
            false
          )
      ))
  }

  handleSurplus(
    pubsub: AtomicPubSub<A>,
    subscribers: Subscribers<A>,
    elements: Iterable<A>,
    isShutdown: MutableRef.MutableRef<boolean>
  ): Effect.Effect<boolean> {
    return core.withFiberRuntime((state) => {
      const deferred = core.deferredUnsafeMake<boolean>(state.id())
      return pipe(
        core.suspend(() => {
          this.unsafeOffer(elements, deferred)
          this.unsafeOnPubSubEmptySpace(pubsub, subscribers)
          this.unsafeCompleteSubscribers(pubsub, subscribers)
          return MutableRef.get(isShutdown) ?
            core.interrupt :
            core.deferredAwait(deferred)
        }),
        core.onInterrupt(() => core.sync(() => this.unsafeRemove(deferred)))
      )
    })
  }

  unsafeOnPubSubEmptySpace(
    pubsub: AtomicPubSub<A>,
    subscribers: Subscribers<A>
  ): void {
    let keepPolling = true
    while (keepPolling && !pubsub.isFull()) {
      const publisher = pipe(this.publishers, MutableQueue.poll(MutableQueue.EmptyMutableQueue))
      if (publisher === MutableQueue.EmptyMutableQueue) {
        keepPolling = false
      } else {
        const published = pubsub.publish(publisher[0])
        if (published && publisher[2]) {
          unsafeCompleteDeferred(publisher[1], true)
        } else if (!published) {