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 Added in v3.10.0 Source

Constructors

make

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

Added in v4.0.0 Source

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

Added in v4.0.0 Source

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/transactions/TxSubscriptionRef": "~effect/transactions/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

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

Added in v3.10.0 Source

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

changesStream for subscribing through a Stream

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

Added in v3.10.0 Source

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]