Skip to content
Effect 4.0 is here. One ecosystem. Zero dependencies.

TxSubscriptionRef

Stores transactional state and publishes committed changes.

A TxSubscriptionRef<A> combines a TxRef<A> for the current value with a transactional pub/sub channel for updates. Subscribers first receive the current value and then every later value that is published by committed updates. This module includes constructors, reads, writes, update and modify helpers, transactional-queue subscriptions, stream subscriptions, and a guard.

12 exports Since v3.10.0 Source

Constructors

make

Since v3.10.0

Creates a new TxSubscriptionRef with the specified initial value.

When to use

Use to create a TxSubscriptionRef that publishes every committed update to subscribers.

See

  • changes for subscribing to the created reference

Signature

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

Example Creating a transactional subscription reference 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(42)
  return yield* TxSubscriptionRef.get(ref)
})


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

Getters

get

Since v3.10.0

Reads the current value of the TxSubscriptionRef.

When to use

Use to read the current TxSubscriptionRef value without subscribing to future changes.

See

  • changes for reading the current value and subsequent updates

Signature

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

Example Reading the current value 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make("hello")
  return yield* TxSubscriptionRef.get(ref)
})


await Effect.runPromise(program) // => "hello"

Guards

isTxSubscriptionRef

Since v4.0.0

Checks whether the given value is a TxSubscriptionRef.

When to use

Use to narrow an unknown value before treating it as a TxSubscriptionRef.

See

  • make for creating a TxSubscriptionRef

Signature

declare function isTxSubscriptionRef(u: unknown): u is TxSubscriptionRef<unknown>

Example Checking transactional subscription references 

import { TxSubscriptionRef } from "effect"


const someValue: unknown = {}
TxSubscriptionRef.isTxSubscriptionRef(someValue) // => false

Models

TxSubscriptionRef

interface
Since v4.0.0

A TxSubscriptionRef is a transactional reference that allows subscribing to all committed changes. Subscribers receive the current value followed by every subsequent update via a transactional dequeue.

When to use

Use to store transactional state whose committed changes must be observable by subscribers.

See

  • make for creating a transactional subscription reference
  • changes for subscribing through a transactional queue
  • changesStream for subscribing through a Stream

Signature

interface TxSubscriptionRef<in out A> extends Inspectable, Pipeable {
  readonly "~effect/TxSubscriptionRef": "~effect/TxSubscriptionRef";
}

Example Subscribing to transactional changes 

import { Effect, TxQueue, TxSubscriptionRef } from "effect"


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


  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxSubscriptionRef.changes(ref)
      const initial = yield* TxQueue.take(sub)


      yield* TxSubscriptionRef.set(ref, 1)
      const next = yield* TxQueue.take(sub)
      return [initial, next]
    })
  )
})


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

Mutations

getAndSet

Since v3.10.0

Gets the current value and sets a new value atomically. Publishes the new value to all subscribers.

When to use

Use to replace a TxSubscriptionRef value while returning the previous value and publishing the update to subscribers.

See

  • set for setting without returning the previous value
  • getAndUpdate for deriving the new value from the previous value

Signature

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

Example Getting and setting atomically 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make("a")
  const old = yield* TxSubscriptionRef.getAndSet(ref, "b")
  return [old, yield* TxSubscriptionRef.get(ref)]
})


await Effect.runPromise(program) // => ["a", "b"]

getAndUpdate

Since v3.10.0

Gets the current value and updates it using a function atomically. Publishes the new value to all subscribers.

When to use

Use to derive and publish a new TxSubscriptionRef value while returning the previous value.

See

  • update for updating without returning the previous value
  • updateAndGet for returning the new value instead

Signature

declare const getAndUpdate: {
  <A>(f: (current: A) => A): (self: TxSubscriptionRef<A>) => Effect<A>;
  <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect<A>;
}

Example Getting and updating atomically 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(1)
  const old = yield* TxSubscriptionRef.getAndUpdate(ref, (n) => n + 10)
  return [old, yield* TxSubscriptionRef.get(ref)]
})


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

modify

Since v3.10.0

Modifies the value of the TxSubscriptionRef using a function that returns both a result and the new value. The new value is published to all subscribers atomically.

When to use

Use to compute a separate return value and next TxSubscriptionRef state in one transactional update.

See

  • update for deriving the next value without a separate return value
  • set for replacing the value directly

Signature

declare const modify: {
  <A, B>(f: (current: A) => [returnValue: B, newValue: A]): (self: TxSubscriptionRef<A>) => Effect<B>;
  <A, B>(self: TxSubscriptionRef<A>, f: (current: A) => [returnValue: B, newValue: A]): Effect<B>;
}

Example Modifying and returning a value 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(10)
  const result = yield* TxSubscriptionRef.modify(ref, (n) => [`was ${n}`, n + 1])
  return [result, yield* TxSubscriptionRef.get(ref)]
})


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

set

Since v3.10.0

Sets the value of the TxSubscriptionRef and publishes the new value to all subscribers.

When to use

Use to replace the current TxSubscriptionRef value with a known value and publish it.

See

  • update for deriving the new value from the current value
  • getAndSet for setting while returning the previous value

Signature

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

Example Setting a new value 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(0)
  yield* TxSubscriptionRef.set(ref, 42)
  return yield* TxSubscriptionRef.get(ref)
})


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

update

Since v3.10.0

Updates the value of the TxSubscriptionRef using a function and publishes the new value to all subscribers.

When to use

Use to derive the next TxSubscriptionRef value from the current value and publish it.

See

  • set for replacing the value directly
  • updateAndGet for returning the new value after the update

Signature

declare const update: {
  <A>(f: (current: A) => A): (self: TxSubscriptionRef<A>) => Effect<void>;
  <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect<void>;
}

Example Updating a value 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(5)
  yield* TxSubscriptionRef.update(ref, (n) => n * 2)
  return yield* TxSubscriptionRef.get(ref)
})


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

updateAndGet

Since v3.10.0

Updates the value using a function and returns the new value. Publishes the new value to all subscribers.

When to use

Use to derive and publish a new TxSubscriptionRef value while returning that new value.

See

  • update for updating without returning the new value
  • getAndUpdate for returning the previous value instead

Signature

declare const updateAndGet: {
  <A>(f: (current: A) => A): (self: TxSubscriptionRef<A>) => Effect<A>;
  <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect<A>;
}

Example Updating and reading atomically 

import { Effect, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(3)
  return yield* TxSubscriptionRef.updateAndGet(ref, (n) => n * 3)
})


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

Subscriptions

changes

unstable
Since v3.10.0

Subscribes to all changes of the TxSubscriptionRef. Returns a scoped TxDequeue that first yields the current value, then every subsequent update.

When to use

Use to subscribe to TxSubscriptionRef committed changes through a scoped transactional queue.

See

Signature

declare function changes<A>(self: TxSubscriptionRef<A>): Effect<TxQueue<A, never>, never, Scope>

Example Subscribing to changes 

import { Effect, TxQueue, TxSubscriptionRef } from "effect"


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


  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxSubscriptionRef.changes(ref)
      const initial = yield* TxQueue.take(sub)


      yield* TxSubscriptionRef.set(ref, 1)
      const next = yield* TxQueue.take(sub)
      return [initial, next]
    })
  )
})


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

changesStream

Since v3.10.0

Returns a Stream of all changes to the TxSubscriptionRef, starting with the current value followed by every subsequent update.

When to use

Use to consume TxSubscriptionRef committed changes as a Stream.

See

  • changes for subscribing through a transactional queue

Signature

declare function changesStream<A>(self: TxSubscriptionRef<A>): Stream<A, never, never>

Example Streaming changes 

import { Effect, Stream, TxSubscriptionRef } from "effect"


const program = Effect.gen(function*() {
  const ref = yield* TxSubscriptionRef.make(0)
  yield* TxSubscriptionRef.set(ref, 1)
  yield* TxSubscriptionRef.set(ref, 2)


  const values = yield* Stream.runCollect(
    TxSubscriptionRef.changesStream(ref).pipe(Stream.take(1))
  )
  return Array.from(values)
})


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