Class SlidingStrategy

packages/effect/src/internal/pubsub.ts:1565–1620

 * @internal
 */
export class SlidingStrategy<in out A> implements PubSubStrategy<A> {
  get shutdown(): Effect.Effect<void> {
    return core.void
  }

  handleSurplus(
    pubsub: AtomicPubSub<A>,
    subscribers: Subscribers<A>,
    elements: Iterable<A>,
    _isShutdown: MutableRef.MutableRef<boolean>
  ): Effect.Effect<boolean> {
    return core.sync(() => {
      this.unsafeSlidingPublish(pubsub, elements)
      this.unsafeCompleteSubscribers(pubsub, subscribers)
      return true
    })
  }

  unsafeOnPubSubEmptySpace(
    _pubsub: AtomicPubSub<A>,
    _subscribers: Subscribers<A>
  ): void {
    //
  }

  unsafeCompletePollers(
    pubsub: AtomicPubSub<A>,
    subscribers: Subscribers<A>,
    subscription: Subscription<A>,
    pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>
  ): void {
    return unsafeStrategyCompletePollers(this, pubsub, subscribers, subscription, pollers)
  }

  unsafeCompleteSubscribers(pubsub: AtomicPubSub<A>, subscribers: Subscribers<A>): void {
    return unsafeStrategyCompleteSubscribers(this, pubsub, subscribers)
  }

  unsafeSlidingPublish(pubsub: AtomicPubSub<A>, elements: Iterable<A>): void {
    const it = elements[Symbol.iterator]()
    let next = it.next()
    if (!next.done && pubsub.capacity > 0) {
      let a = next.value
      let loop = true
      while (loop) {
        pubsub.slide()
        const pub = pubsub.publish(a)
        if (pub && (next = it.next()) && !next.done) {
          a = next.value
        } else if (pub) {
          loop = false
        }
      }
    }
  }
}

/** @internal */