Class UnboundedPubSub

repos/effect/packages/effect/src/PubSub.ts:2065–2146

    this.replayBuffer = replayBuffer
  }

  replayWindow(): PubSub.ReplayWindow<A> {
    return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
  }

  isEmpty(): boolean {
    return this.publisherHead === this.publisherTail
  }

  isFull(): boolean {
    return false
  }

  size(): number {
    return this.publisherIndex - this.subscribersIndex
  }

  publish(value: A): boolean {
    const replayIndex = this.replayBuffer?.offer(value)
    const subscribers = this.publisherTail.subscribers
    if (subscribers !== 0) {
      const node: Node<A> = {
        value,
        replayIndex,
        subscribers,
        next: null
      }
      this.publisherTail.next = node
      this.publisherTail = this.publisherTail.next
      this.publisherIndex += 1
    }
    return true
  }

  publishAll(elements: Iterable<A>): Array<A> {
    if (this.publisherTail.subscribers !== 0) {
      for (const a of elements) {
        this.publish(a)
      }
    } else if (this.replayBuffer) {
      this.replayBuffer.offerAll(elements)
    }
    return []
  }

  slide(): void {
    if (this.publisherHead !== this.publisherTail) {
      const node = this.publisherHead.next!
      const value = node.value as A
      this.publisherHead = this.publisherHead.next!
      this.publisherHead.value = AbsentValue
      this.subscribersIndex += 1
      this.replayBuffer?.slide(value, node.replayIndex!)
    }
  }

  subscribe(): PubSub.BackingSubscription<A> {
    this.publisherTail.subscribers += 1