Class PubSubImpl

packages/effect/src/internal/pubsub.ts:1115–1256

@internal

/** @internal */
class PubSubImpl<in out A> implements PubSub.PubSub<A> {
  readonly [queue.EnqueueTypeId] = queue.enqueueVariance
  readonly [queue.DequeueTypeId] = queue.dequeueVariance

  constructor(
    readonly pubsub: AtomicPubSub<A>,
    readonly subscribers: Subscribers<A>,
    readonly scope: Scope.Scope.Closeable,
    readonly shutdownHook: Deferred.Deferred<void>,
    readonly shutdownFlag: MutableRef.MutableRef<boolean>,
    readonly strategy: PubSubStrategy<A>
  ) {}

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

  get size(): Effect.Effect<number> {
    return core.suspend(() =>
      MutableRef.get(this.shutdownFlag) ?
        core.interrupt :
        core.sync(() => this.pubsub.size())
    )
  }

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

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

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

  get awaitShutdown(): Effect.Effect<void> {
    return core.deferredAwait(this.shutdownHook)
  }

  get isShutdown(): Effect.Effect<boolean> {
    return core.sync(() => MutableRef.get(this.shutdownFlag))
  }

  get shutdown(): Effect.Effect<void> {
    return core.uninterruptible(core.withFiberRuntime((state) => {
      pipe(this.shutdownFlag, MutableRef.set(true))
      return pipe(
        this.scope.close(core.exitInterrupt(state.id())),
        core.zipRight(this.strategy.shutdown),
        core.whenEffect(core.deferredSucceed(this.shutdownHook, void 0)),
        core.asVoid
      )
    }))