Class TSubscriptionRefImpl

packages/effect/src/internal/stm/tSubscriptionRef.ts:111–153

@internal

/** @internal */
class TSubscriptionRefImpl<in out A> implements TSubscriptionRef.TSubscriptionRef<A> {
  readonly [TSubscriptionRefTypeId] = TSubscriptionRefVariance
  readonly [TRef.TRefTypeId] = tRefVariance

  constructor(
    readonly ref: TRef.TRef<A>,
    readonly pubsub: TPubSub.TPubSub<A>
  ) {}

  get todos() {
    return this.ref.todos
  }

  get versioned() {
    return this.ref.versioned
  }

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

  get changes(): STM.STM<TQueue.TDequeue<A>> {
    return STM.gen(this, function*() {
      const first = yield* TQueue.unbounded<A>()
      yield* TQueue.offer(first, yield* TRef.get(this.ref))
      return new TDequeueMerge(first, yield* TPubSub.subscribe(this.pubsub))
    })
  }

  modify<B>(f: (a: A) => readonly [B, A]): STM.STM<B> {
    return pipe(
      TRef.get(this.ref),
      STM.map(f),
      STM.flatMap(([b, a]) =>
        pipe(
          TRef.set(this.ref, a),
          STM.as(b),
          STM.zipLeft(TPubSub.publish(this.pubsub, a))
        )
      )
    )
  }
}

/** @internal */
export const make = <A>(value: A): STM.STM<TSubscriptionRef.TSubscriptionRef<A>> =>