Class QueueImpl

packages/effect/src/internal/queue.ts:67–304

@internal

/** @internal */
class QueueImpl<in out A> extends Effectable.Class<A> implements Queue.Queue<A> {
  readonly [EnqueueTypeId] = enqueueVariance
  readonly [DequeueTypeId] = dequeueVariance

  constructor(
    /** @internal */
    readonly queue: Queue.BackingQueue<A>,
    /** @internal */
    readonly takers: MutableQueue.MutableQueue<Deferred.Deferred<A>>,
    /** @internal */
    readonly shutdownHook: Deferred.Deferred<void>,
    /** @internal */
    readonly shutdownFlag: MutableRef.MutableRef<boolean>,
    /** @internal */
    readonly strategy: Queue.Strategy<A>
  ) {
    super()
  }

  pipe() {
    return pipeArguments(this, arguments)
  }

  commit() {
    return this.take
  }

  capacity(): number {
    return this.queue.capacity()
  }

  get size(): Effect.Effect<number> {
    return core.suspend(() => core.catchAll(this.unsafeSize(), () => core.interrupt))
  }

  unsafeSize() {
    if (MutableRef.get(this.shutdownFlag)) {
      return Option.none<number>()
    }
    return Option.some(
      this.queue.length() -
        MutableQueue.length(this.takers) +
        this.strategy.surplusSize()
    )
  }

  get isEmpty(): Effect.Effect<boolean> {
    return core.map(this.size, (size) => size <= 0)
  }

  get isFull(): Effect.Effect<boolean> {
    return core.map(this.size, (size) => size >= this.capacity())
  }

  get shutdown(): Effect.Effect<void> {
    return core.uninterruptible(
      core.withFiberRuntime((state) => {
        pipe(this.shutdownFlag, MutableRef.set(true))