Method take

packages/effect/src/internal/pubsub.ts:1002–1035

()

  }

  get take(): Effect.Effect<A> {
    return core.withFiberRuntime((state) => {
      if (MutableRef.get(this.shutdownFlag)) {
        return core.interrupt
      }
      if (this.replayWindow.remaining > 0) {
        const message = this.replayWindow.take()!
        return core.succeed(message)
      }
      const message = MutableQueue.isEmpty(this.pollers)
        ? this.subscription.poll(MutableQueue.EmptyMutableQueue)
        : MutableQueue.EmptyMutableQueue
      if (message === MutableQueue.EmptyMutableQueue) {
        const deferred = core.deferredUnsafeMake<A>(state.id())
        return pipe(
          core.suspend(() => {
            pipe(this.pollers, MutableQueue.offer(deferred))
            pipe(this.subscribers, addSubscribers(this.subscription, this.pollers))
            this.strategy.unsafeCompletePollers(
              this.pubsub,
              this.subscribers,
              this.subscription,
              this.pollers
            )
            return MutableRef.get(this.shutdownFlag) ? core.interrupt : core.deferredAwait(deferred)
          }),
          core.onInterrupt(() => core.sync(() => unsafeRemove(this.pollers, deferred)))
        )
      } else {
        this.strategy.unsafeOnPubSubEmptySpace(this.pubsub, this.subscribers)
        return core.succeed(message)
      }
    })
  }

  get takeAll(): Effect.Effect<Chunk.Chunk<A>> {
    return core.suspend(() => {