Class SubscriptionRefImpl

packages/effect/src/internal/subscriptionRef.ts:29–81

@internal

/** @internal */
class SubscriptionRefImpl<in out A> extends Effectable.Class<A> implements SubscriptionRef.SubscriptionRef<A> {
  readonly [Readable.TypeId]: Readable.TypeId = Readable.TypeId
  readonly [Subscribable.TypeId]: Subscribable.TypeId = Subscribable.TypeId
  readonly [Ref.RefTypeId] = ref_.refVariance
  readonly [Synchronized.SynchronizedRefTypeId] = circular_.synchronizedVariance
  readonly [SubscriptionRefTypeId] = subscriptionRefVariance
  constructor(
    readonly ref: Ref.Ref<A>,
    readonly pubsub: PubSub.PubSub<A>,
    readonly semaphore: Effect.Semaphore
  ) {
    super()
    this.get = Ref.get(this.ref)
  }
  commit() {
    return this.get
  }
  readonly get: Effect.Effect<A>
  get changes(): Stream<A> {
    return pipe(
      Ref.get(this.ref),
      Effect.flatMap((a) =>
        Effect.map(
          stream.fromPubSub(this.pubsub, { scoped: true }),
          (s) =>
            stream.concat(
              stream.make(a),
              s
            )
        )
      ),
      this.semaphore.withPermits(1),
      stream.unwrapScoped
    )
  }
  modify<B>(f: (a: A) => readonly [B, A]): Effect.Effect<B> {
    return this.modifyEffect((a) => Effect.succeed(f(a)))
  }
  modifyEffect<B, E, R>(f: (a: A) => Effect.Effect<readonly [B, A], E, R>): Effect.Effect<B, E, R> {
    return pipe(
      Ref.get(this.ref),
      Effect.flatMap(f),
      Effect.flatMap(([b, a]) =>
        pipe(
          Ref.set(this.ref, a),
          Effect.as(b),
          Effect.zipLeft(PubSub.publish(this.pubsub, a))
        )
      ),
      this.semaphore.withPermits(1)
    )
  }
}

/** @internal */
export const get = <A>(self: SubscriptionRef.SubscriptionRef<A>): Effect.Effect<A> => Ref.get(self.ref)