SubscriptionRef

Stores mutable state and publishes changes as a stream.

A SubscriptionRef<A> stores the latest value, publishes the initial value, and publishes every committed update so subscribers can observe state over time. Updates are serialized so only one change is applied at a time. This module includes constructors, current-value reads, the changes stream, writes, updates, partial updates, and effectful update helpers.

26 exports Since v2.0.0 Source

Constructors

make

Since v2.0.0

Constructs a new SubscriptionRef from an initial value.

When to use

Use to create a SubscriptionRef when consumers need to read the latest value and subscribe to every update.

Details

The initial value is published during construction, so changes starts new subscribers with that value before future updates.

See

changes for streaming the current value and subsequent updates
set for replacing the value and notifying subscribers

Signature

declare function make<A>(value: A): Effect<SubscriptionRef<A>>

Getters

get

Since v2.0.0

Retrieves the current value of the SubscriptionRef.

Signature

declare function get<A>(self: SubscriptionRef<A>): Effect<A>

Example Reading the current value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(42)

  return yield* SubscriptionRef.get(ref)
})

await Effect.runPromise(program) // => 42

getAndSet

Since v2.0.0

Retrieves the current value and sets a new value atomically, notifying subscribers of the change.

Signature

declare const getAndSet: {
  <A>(value: A): (self: SubscriptionRef<A>) => Effect<A>;
  <A>(self: SubscriptionRef<A>, value: A): Effect<A>;
}

Example Getting and setting a value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const oldValue = yield* SubscriptionRef.getAndSet(ref, 20)
  const newValue = yield* SubscriptionRef.get(ref)
  return [oldValue, newValue]
})

await Effect.runPromise(program) // => [10, 20]

getAndUpdate

Since v2.0.0

Retrieves the current value and updates it atomically with the result of applying a function, notifying subscribers of the change.

Signature

declare const getAndUpdate: {
  <A>(update: (a: A) => A): (self: SubscriptionRef<A>) => Effect<A>;
  <A>(self: SubscriptionRef<A>, update: (a: A) => A): Effect<A>;
}

Example Getting and updating a value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const oldValue = yield* SubscriptionRef.getAndUpdate(ref, (n) => n * 2)
  const newValue = yield* SubscriptionRef.get(ref)
  return [oldValue, newValue]
})

await Effect.runPromise(program) // => [10, 20]

getAndUpdateEffect

Since v2.0.0

Retrieves the current value and updates it atomically with the result of applying an effectful function, notifying subscribers of the change.

Signature

declare const getAndUpdateEffect: {
  <A, E, R>(update: (a: A) => Effect<A, E, R>): (self: SubscriptionRef<A>) => Effect<A, E, R>;
  <A, E, R>(self: SubscriptionRef<A>, update: (a: A) => Effect<A, E, R>): Effect<A, E, R>;
}

Example Getting and updating with an effect

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const oldValue = yield* SubscriptionRef.getAndUpdateEffect(
    ref,
    (n) => Effect.succeed(n + 5)
  )
  const newValue = yield* SubscriptionRef.get(ref)
  return [oldValue, newValue]
})

await Effect.runPromise(program) // => [10, 15]

getAndUpdateSome

Since v2.0.0

Retrieves the current value and optionally updates the reference.

When to use

Use to read the old SubscriptionRef value while applying a synchronous update only when a new value is available.

Details

If the function returns Option.some, the new value is set and published. If it returns Option.none, the reference is left unchanged and no update is published.

Signature

declare const getAndUpdateSome: {
  <A>(update: (a: A) => Option<A>): (self: SubscriptionRef<A>) => Effect<A>;
  <A>(self: SubscriptionRef<A>, update: (a: A) => Option<A>): Effect<A>;
}

Example Getting and conditionally updating a value

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const oldValue = yield* SubscriptionRef.getAndUpdateSome(
    ref,
    (n) => n > 5 ? Option.some(n * 2) : Option.none()
  )
  const newValue = yield* SubscriptionRef.get(ref)
  return [oldValue, newValue]
})

await Effect.runPromise(program) // => [10, 20]

getAndUpdateSomeEffect

Since v2.0.0

Retrieves the current value and optionally updates the reference effectfully.

When to use

Use to read the old SubscriptionRef value while applying an effectful update only when a new value is available.

Details

If the effect succeeds with Option.some, the new value is set and published. If it succeeds with Option.none, the reference is left unchanged and no update is published.

Signature

declare const getAndUpdateSomeEffect: {
  <A, R, E>(update: (a: A) => Effect<Option<A>, E, R>): (self: SubscriptionRef<A>) => Effect<A, E, R>;
  <A, R, E>(self: SubscriptionRef<A>, update: (a: A) => Effect<Option<A>, E, R>): Effect<A, E, R>;
}

Example Getting and conditionally updating with an effect

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const oldValue = yield* SubscriptionRef.getAndUpdateSomeEffect(
    ref,
    (n) => Effect.succeed(n > 5 ? Option.some(n + 3) : Option.none())
  )
  const newValue = yield* SubscriptionRef.get(ref)
  return [oldValue, newValue]
})

await Effect.runPromise(program) // => [10, 13]

getUnsafe

Since v4.0.0

Retrieves the current value of the SubscriptionRef unsafely.

When to use

Use when you are in synchronous internals or test setup where concurrent updates are controlled.

Gotchas

This function directly accesses the underlying reference without any synchronization. It should only be used when you are certain there are no concurrent modifications.

Signature

declare function getUnsafe<A>(self: SubscriptionRef<A>): A

Example Reading the current value unsafely

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(42)

  return SubscriptionRef.getUnsafe(ref)
})

await Effect.runPromise(program) // => 42

Guards

isSubscriptionRef

Since v4.0.0

Returns true if the provided value is a SubscriptionRef.

When to use

Use to narrow an unknown value before calling SubscriptionRef operations that require a subscription reference.

Signature

declare const isSubscriptionRef: (u: unknown) => u is SubscriptionRef<unknown>

Models

SubscriptionRef

interface

Since v2.0.0

A mutable reference whose updates are serialized and published to subscribers.

When to use

Use to observe the current value and subsequent updates as a stream.

Signature

interface SubscriptionRef<in out A> extends Variance<A>, Pipeable {
  readonly pubsub: PubSub<A>;
  readonly semaphore: Semaphore;
  value: A;
}

Mutations

modify

Since v2.0.0

Modifies the SubscriptionRef atomically with a function that computes a return value and a new value, notifying subscribers of the change.

Signature

declare const modify: {
  <A, B>(modify: (a: A) => readonly [B, A]): (self: SubscriptionRef<A>) => Effect<B>;
  <A, B>(self: SubscriptionRef<A>, f: (a: A) => readonly [B, A]): Effect<B>;
}

Example Modifying a value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const result = yield* SubscriptionRef.modify(ref, (n) => [
    `Old value was ${n}`,
    n * 2
  ])
  const newValue = yield* SubscriptionRef.get(ref)
  return [result, newValue]
})

await Effect.runPromise(program) // => ["Old value was 10", 20]

modifyEffect

Since v2.0.0

Modifies the SubscriptionRef atomically with an effectful function that computes a return value and a new value, notifying subscribers of the change.

Signature

declare const modifyEffect: {
  <B, A, E, R>(modify: (a: A) => Effect<readonly [B, A], E, R>): (self: SubscriptionRef<A>) => Effect<B, E, R>;
  <A, B, E, R>(self: SubscriptionRef<A>, modify: (a: A) => Effect<readonly [B, A], E, R>): Effect<B, E, R>;
}

Example Modifying with an effect

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const result = yield* SubscriptionRef.modifyEffect(
    ref,
    (n) => Effect.succeed([`Doubled from ${n}`, n * 2] as const)
  )
  const newValue = yield* SubscriptionRef.get(ref)
  return [result, newValue]
})

await Effect.runPromise(program) // => ["Doubled from 10", 20]

modifySome

Since v2.0.0

Computes a return value and optionally updates the reference.

When to use

Use to return a separate result while synchronously deciding whether to publish a new SubscriptionRef value.

Details

If the function returns Option.some for the new value, the value is set and published. If it returns Option.none, the reference is left unchanged and no update is published.

Signature

declare const modifySome: {
  <B, A>(modify: (a: A) => readonly [B, Option<A>]): (self: SubscriptionRef<A>) => Effect<B>;
  <A, B>(self: SubscriptionRef<A>, modify: (a: A) => readonly [B, Option<A>]): Effect<B>;
}

Example Conditionally modifying a value

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const result = yield* SubscriptionRef.modifySome(
    ref,
    (n) =>
      n > 5 ? ["Updated", Option.some(n * 2)] : ["Not updated", Option.none()]
  )
  const newValue = yield* SubscriptionRef.get(ref)
  return [result, newValue]
})

await Effect.runPromise(program) // => ["Updated", 20]

modifySomeEffect

Since v2.0.0

Computes a return value and optionally updates the reference effectfully.

When to use

Use to return a separate result while effectfully deciding whether to publish a new SubscriptionRef value.

Details

If the effect succeeds with Option.some, the new value is set and published. If it succeeds with Option.none, the reference is left unchanged and no update is published.

Signature

declare const modifySomeEffect: {
  <A, B, R, E>(modify: (a: A) => Effect<readonly [B, Option<A>], E, R>): (self: SubscriptionRef<A>) => Effect<B, E, R>;
  <A, B, R, E>(self: SubscriptionRef<A>, modify: (a: A) => Effect<readonly [B, Option<A>], E, R>): Effect<B, E, R>;
}

Example Conditionally modifying with an effect

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  const result = yield* SubscriptionRef.modifySomeEffect(
    ref,
    (n) =>
      Effect.succeed(
        n > 5
          ? (["Updated", Option.some(n + 5)] as const)
          : (["Not updated", Option.none()] as const)
      )
  )
  const newValue = yield* SubscriptionRef.get(ref)
  return [result, newValue]
})

await Effect.runPromise(program) // => ["Updated", 15]

set

Since v2.0.0

Sets the value of the SubscriptionRef, notifying all subscribers of the change.

Signature

declare const set: {
  <A>(value: A): (self: SubscriptionRef<A>) => Effect<void>;
  <A>(self: SubscriptionRef<A>, value: A): Effect<void>;
}

Example Setting a value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(0)

  yield* SubscriptionRef.set(ref, 42)

  return yield* SubscriptionRef.get(ref)
})

await Effect.runPromise(program) // => 42

setAndGet

Since v2.0.0

Sets the value of the SubscriptionRef and returns the new value, notifying all subscribers of the change.

Signature

declare const setAndGet: {
  <A>(value: A): (self: SubscriptionRef<A>) => Effect<A>;
  <A>(self: SubscriptionRef<A>, value: A): Effect<A>;
}

Example Setting and reading the new value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(0)

  return yield* SubscriptionRef.setAndGet(ref, 42)
})

await Effect.runPromise(program) // => 42

update

Since v2.0.0

Updates the value of the SubscriptionRef with the result of applying a function, notifying subscribers of the change.

Signature

declare const update: {
  <A>(update: (a: A) => A): (self: SubscriptionRef<A>) => Effect<void>;
  <A>(self: SubscriptionRef<A>, update: (a: A) => A): Effect<void>;
}

Example Updating a value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  yield* SubscriptionRef.update(ref, (n) => n * 2)

  return yield* SubscriptionRef.get(ref)
})

await Effect.runPromise(program) // => 20

updateAndGet

Since v2.0.0

Updates the value of the SubscriptionRef with the result of applying a function and returns the new value, notifying subscribers of the change.

Signature

declare const updateAndGet: {
  <A>(update: (a: A) => A): (self: SubscriptionRef<A>) => Effect<A>;
  <A>(self: SubscriptionRef<A>, update: (a: A) => A): Effect<A>;
}

Example Updating and reading the new value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  return yield* SubscriptionRef.updateAndGet(ref, (n) => n * 2)
})

await Effect.runPromise(program) // => 20

updateAndGetEffect

Since v2.0.0

Updates the value of the SubscriptionRef with the result of applying an effectful function and returns the new value, notifying subscribers of the change.

Signature

declare const updateAndGetEffect: {
  <A, E, R>(update: (a: A) => Effect<A, E, R>): (self: SubscriptionRef<A>) => Effect<A, E, R>;
  <A, E, R>(self: SubscriptionRef<A>, update: (a: A) => Effect<A, E, R>): Effect<A, E, R>;
}

Example Updating with an effect and reading the new value

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  return yield* SubscriptionRef.updateAndGetEffect(
    ref,
    (n) => Effect.succeed(n + 5)
  )
})

await Effect.runPromise(program) // => 15

updateEffect

Since v2.0.0

Updates the value of the SubscriptionRef with the result of applying an effectful function, notifying subscribers of the change.

Signature

declare const updateEffect: {
  <A, E, R>(update: (a: A) => Effect<A, E, R>): (self: SubscriptionRef<A>) => Effect<void, E, R>;
  <A, E, R>(self: SubscriptionRef<A>, update: (a: A) => Effect<A, E, R>): Effect<void, E, R>;
}

Example Updating with an effect

import { Effect, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  yield* SubscriptionRef.updateEffect(ref, (n) => Effect.succeed(n + 5))

  return yield* SubscriptionRef.get(ref)
})

await Effect.runPromise(program) // => 15

updateSome

Since v2.0.0

Applies an update function to the current value. If it returns Option.some, sets and publishes that value; if it returns Option.none, leaves the reference unchanged and does not publish.

Signature

declare const updateSome: {
  <A>(update: (a: A) => Option<A>): (self: SubscriptionRef<A>) => Effect<void>;
  <A>(self: SubscriptionRef<A>, update: (a: A) => Option<A>): Effect<void>;
}

Example Conditionally updating a value

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  yield* SubscriptionRef.updateSome(
    ref,
    (n) => n > 5 ? Option.some(n * 2) : Option.none()
  )

  return yield* SubscriptionRef.get(ref)
})

await Effect.runPromise(program) // => 20

updateSomeAndGet

Since v2.0.0

Applies an optional update and returns the current value afterward.

When to use

Use to conditionally update a SubscriptionRef and read the value that is current after the update decision.

Details

If the function returns Option.some, the new value is set, published, and returned. If it returns Option.none, the unchanged current value is returned without publishing.

Signature

declare const updateSomeAndGet: {
  <A>(update: (a: A) => Option<A>): (self: SubscriptionRef<A>) => Effect<A>;
  <A>(self: SubscriptionRef<A>, update: (a: A) => Option<A>): Effect<A>;
}

Example Conditionally updating and reading the new value

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  return yield* SubscriptionRef.updateSomeAndGet(
    ref,
    (n) => n > 5 ? Option.some(n * 2) : Option.none()
  )
})

await Effect.runPromise(program) // => 20

updateSomeAndGetEffect

Since v2.0.0

Applies an effectful optional update and returns the current value afterward.

When to use

Use to conditionally update a SubscriptionRef effectfully and read the value that is current after the update decision.

Details

If the effect succeeds with Option.some, the new value is set, published, and returned. If it succeeds with Option.none, the unchanged current value is returned without publishing.

Signature

declare const updateSomeAndGetEffect: {
  <A, E, R>(update: (a: A) => Effect<Option<A>, E, R>): (self: SubscriptionRef<A>) => Effect<A, E, R>;
  <A, E, R>(self: SubscriptionRef<A>, update: (a: A) => Effect<Option<A>, E, R>): Effect<A, E, R>;
}

Example Conditionally updating with an effect and reading the new value

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  return yield* SubscriptionRef.updateSomeAndGetEffect(
    ref,
    (n) => Effect.succeed(n > 5 ? Option.some(n + 3) : Option.none())
  )
})

await Effect.runPromise(program) // => 13

updateSomeEffect

Since v2.0.0

Applies an effectful update only when it produces a new value.

When to use

Use to conditionally update a SubscriptionRef with an effectful function while discarding the resulting value.

Details

If the effect succeeds with Option.some, the new value is set and published. If it succeeds with Option.none, the reference is left unchanged and no update is published.

Signature

declare const updateSomeEffect: {
  <A, E, R>(update: (a: A) => Effect<Option<A>, E, R>): (self: SubscriptionRef<A>) => Effect<void, E, R>;
  <A, E, R>(self: SubscriptionRef<A>, update: (a: A) => Effect<Option<A>, E, R>): Effect<void, E, R>;
}

Example Conditionally updating with an effect

import { Effect, Option, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(10)

  yield* SubscriptionRef.updateSomeEffect(
    ref,
    (n) => Effect.succeed(n > 5 ? Option.some(n + 3) : Option.none())
  )

  return yield* SubscriptionRef.get(ref)
})

await Effect.runPromise(program) // => 13

Details

The stream will first emit the current value, then emit all future changes as they occur.

Signature

declare function changes<A>(self: SubscriptionRef<A>): Stream<A>

Example Streaming changes

import { Deferred, Effect, Fiber, Stream, SubscriptionRef } from "effect"

const program = Effect.gen(function*() {
  const ref = yield* SubscriptionRef.make(0)
  const ready = yield* Deferred.make<void>()

  const fiber = yield* SubscriptionRef.changes(ref).pipe(
    Stream.tap(() => Deferred.succeed(ready, void 0)),
    Stream.take(3),
    Stream.runCollect,
    Effect.forkChild
  )

  yield* Deferred.await(ready)
  yield* SubscriptionRef.set(ref, 1)
  yield* SubscriptionRef.set(ref, 2)

  const values = yield* Fiber.join(fiber)
  return Array.from(values)
})

await Effect.runPromise(program) // => [0, 1, 2]