Class SubscriptionImpl

packages/effect/src/internal/pubsub.ts:913–1077

@internal

/** @internal */
class SubscriptionImpl<in out A> extends Effectable.Class<A> implements Queue.Dequeue<A> {
  [queue.DequeueTypeId] = queue.dequeueVariance

  constructor(
    readonly pubsub: AtomicPubSub<A>,
    readonly subscribers: Subscribers<A>,
    readonly subscription: Subscription<A>,
    readonly pollers: MutableQueue.MutableQueue<Deferred.Deferred<A>>,
    readonly shutdownHook: Deferred.Deferred<void>,
    readonly shutdownFlag: MutableRef.MutableRef<boolean>,
    readonly strategy: PubSubStrategy<A>,
    readonly replayWindow: ReplayWindow<A>
  ) {
    super()
  }

  commit() {
    return this.take
  }

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

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

  isActive(): boolean {
    return !MutableRef.get(this.shutdownFlag)
  }

  get size(): Effect.Effect<number> {
    return core.suspend(() =>
      MutableRef.get(this.shutdownFlag)
        ? core.interrupt
        : core.succeed(this.subscription.size() + this.replayWindow.remaining)
    )
  }

  unsafeSize(): Option.Option<number> {
    if (MutableRef.get(this.shutdownFlag)) {
      return Option.none()
    }
    return Option.some(this.subscription.size() + this.replayWindow.remaining)
  }

  get isFull(): Effect.Effect<boolean> {
    return core.suspend(() =>
      MutableRef.get(this.shutdownFlag)
        ? core.interrupt
        : core.succeed(this.subscription.size() === this.capacity())
    )
  }

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