TxPubSub

Broadcasts values to subscribers inside Effect transactions.

A TxPubSub<A> is a transactional publish/subscribe hub. Each subscriber owns a TxQueue, and each published value is offered to the subscriber queues that are registered at the time of publication. This module includes bounded, dropping, sliding, and unbounded hubs, publishing helpers, scoped subscriptions, shutdown operations, and a guard.

18 exports Added in v2.0.0 Source

Constructors

bounded

Added in v2.0.0 Source

Creates a bounded TxPubSub with the specified capacity. When a subscriber's queue is full, the publisher will retry the transaction until space is available.

Signature

declare function bounded<A = never>(capacity: number): Effect<TxPubSub<A>>

Example

(Creating a bounded pub/sub)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.bounded<number>(16)

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, 42)
      return yield* TxQueue.take(sub)
    })
  )
})

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

dropping

Added in v2.0.0 Source

Creates a dropping TxPubSub with the specified capacity. When a subscriber's queue is full, the message is dropped for that subscriber.

Signature

declare function dropping<A = never>(capacity: number): Effect<TxPubSub<A>>

Example

(Creating a dropping pub/sub)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.dropping<number>(2)

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, 1)
      yield* TxPubSub.publish(hub, 2)
      yield* TxPubSub.publish(hub, 3) // dropped
      const v1 = yield* TxQueue.take(sub)
      const v2 = yield* TxQueue.take(sub)
      return [v1, v2]
    })
  )
})

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

sliding

Added in v2.0.0 Source

Creates a sliding TxPubSub with the specified capacity. When a subscriber's queue is full, the oldest message in that subscriber's queue is dropped.

Signature

declare function sliding<A = never>(capacity: number): Effect<TxPubSub<A>>

Example

(Creating a sliding pub/sub)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.sliding<number>(2)

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, 1)
      yield* TxPubSub.publish(hub, 2)
      yield* TxPubSub.publish(hub, 3) // evicts 1
      return yield* TxQueue.take(sub)
    })
  )
})

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

unbounded

Added in v2.0.0 Source

Creates an unbounded TxPubSub with unlimited capacity. Messages are always accepted.

Signature

declare function unbounded<A = never>(): Effect<TxPubSub<A>>

Example

(Creating an unbounded pub/sub)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<string>()

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, "msg")
      return yield* TxQueue.take(sub)
    })
  )
})

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

Getters

capacity

Added in v2.0.0 Source

Returns the capacity of the TxPubSub.

Signature

declare function capacity<A>(self: TxPubSub<A>): number

Example

(Reading pub/sub capacity)

import { Effect, TxPubSub } from "effect"

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.bounded<number>(16)
  return TxPubSub.capacity(hub)
})

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

size

Added in v2.0.0 Source

Returns the current number of messages across all subscriber queues (the max).

Signature

declare function size<A>(self: TxPubSub<A>): Effect<number>

Example

(Reading subscriber queue size)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<number>()

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, 1)
      yield* TxPubSub.publish(hub, 2)
      return yield* TxPubSub.size(hub)
    })
  )
})

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

Guards

isTxPubSub

Added in v4.0.0 Source

Checks whether the given value is a TxPubSub.

Signature

declare function isTxPubSub(u: unknown): u is TxPubSub<unknown>

Example

(Checking for a TxPubSub)

import { TxPubSub } from "effect"

const someValue: unknown = {}
TxPubSub.isTxPubSub(someValue) // => false

Models

TxPubSub interface

Added in v4.0.0 Source

A TxPubSub represents a transactional publish/subscribe hub that broadcasts messages to all current subscribers using Software Transactional Memory (STM) semantics.

Signature

interface TxPubSub<in out A> extends Inspectable, Pipeable {
  readonly "~effect/transactions/TxPubSub": "~effect/transactions/TxPubSub";
  readonly capacity: number;
  readonly strategy: "sliding" | "dropping" | "unbounded" | "bounded";
}

Example

(Subscribing to a transactional pub/sub)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<string>()

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, "hello")
      return yield* TxQueue.take(sub)
    })
  )
})

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

Mutations

acquireSubscriber

Added in v4.0.0 Source

Creates a subscriber queue and registers it with the pub/sub.

When to use

Use to create and register a subscriber queue inside a larger transaction when registration must be atomic with other Tx operations.

Details

This is the transactional acquire step of subscribe, exposed so that callers can compose it with other Tx operations in a single transaction, such as TxSubscriptionRef.changes.

See

subscribe for the scoped acquire and release wrapper when no custom transaction composition is needed
releaseSubscriber to remove and shut down a queue returned by acquireSubscriber

Signature

declare function acquireSubscriber<A>(self: TxPubSub<A>): Effect<TxQueue<A, never>, never, Transaction>

awaitShutdown

Added in v2.0.0 Source

Waits for the TxPubSub to be shut down.

Signature

declare function awaitShutdown<A>(self: TxPubSub<A>): Effect<void>

Example

(Waiting for shutdown)

import { Effect, Fiber, TxPubSub } from "effect"

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<number>()

  const fiber = yield* Effect.forkChild(TxPubSub.awaitShutdown(hub))
  yield* TxPubSub.shutdown(hub)
  yield* Fiber.await(fiber)
  return yield* TxPubSub.isShutdown(hub)
})

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

publish

Added in v2.0.0 Source

Publishes a message to all current subscribers.

Details

Returns true if the message was delivered to all subscribers, or false if the hub is shut down or the message was dropped for any subscriber. For the bounded strategy, the transaction retries if any subscriber queue is full. For the sliding strategy, full subscriber queues drop their oldest messages. For the dropping strategy, full subscriber queues drop the new message and the operation returns false.

Signature

declare const publish: {
  <A>(value: A): (self: TxPubSub<A>) => Effect<boolean>;
  <A>(self: TxPubSub<A>, value: A): Effect<boolean>;
}

Example

(Publishing a message to subscribers)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<string>()

  // No subscribers - publish is a no-op
  const r1 = yield* TxPubSub.publish(hub, "no one listening")

  const msg = yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publish(hub, "hello")
      return yield* TxQueue.take(sub)
    })
  )
  return [r1, msg]
})

await Effect.runPromise(program) // => [true, "hello"]

publishAll

Added in v2.0.0 Source

Publishes all messages from an iterable to all current subscribers.

Details

Returns true if all messages were delivered to all subscribers.

Signature

declare const publishAll: {
  <A>(values: Iterable<A>): (self: TxPubSub<A>) => Effect<boolean>;
  <A>(self: TxPubSub<A>, values: Iterable<A>): Effect<boolean>;
}

Example

(Publishing multiple messages to subscribers)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<number>()

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub = yield* TxPubSub.subscribe(hub)
      yield* TxPubSub.publishAll(hub, [1, 2, 3])
      const v1 = yield* TxQueue.take(sub)
      const v2 = yield* TxQueue.take(sub)
      const v3 = yield* TxQueue.take(sub)
      return [v1, v2, v3]
    })
  )
})

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

releaseSubscriber

Added in v4.0.0 Source

Removes a subscriber queue from the pub/sub and shuts it down.

When to use

Use to release a manually acquired subscriber queue inside a larger transaction, removing it from the pub/sub and shutting it down together with related transactional cleanup.

Details

This is the transactional release step of subscribe, exposed so that callers can compose it with other Tx operations in a single transaction.

Gotchas

The supplied queue is shut down after being removed, so callers should pass a queue acquired for this pub/sub.

See

acquireSubscriber for the matching transactional acquire step
subscribe for the scoped acquire and release wrapper

Signature

declare const releaseSubscriber: {
  <A>(queue: TxQueue<A>): (self: TxPubSub<A>) => Effect<void, never, Transaction>;
  <A>(self: TxPubSub<A>, queue: TxQueue<A>): Effect<void, never, Transaction>;
}

shutdown

Added in v2.0.0 Source

Shuts down the TxPubSub and all subscriber queues registered at the time of shutdown.

Details

After shutdown, publish and publishAll return false, and awaitShutdown completes. The operation is idempotent.

Gotchas

Subscribers acquired after shutdown are not automatically shut down by this call.

Signature

declare function shutdown<A>(self: TxPubSub<A>): Effect<void>

Example

(Shutting down a pub/sub)

import { Effect, TxPubSub } from "effect"

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<number>()
  yield* TxPubSub.shutdown(hub)

  const shut = yield* TxPubSub.isShutdown(hub)
  const accepted = yield* TxPubSub.publish(hub, 1)
  return [shut, accepted]
})

await Effect.runPromise(program) // => [true, false]

Added in v2.0.0 Source

Subscribes to the TxPubSub, returning a scoped TxQueue for messages published after subscription.

Details

The returned queue uses the hub's capacity strategy: bounded subscriptions backpressure publishers when full, dropping subscriptions may miss new messages when full, and sliding subscriptions may evict older queued messages. The subscription is automatically removed when the scope is closed.

Signature

declare function subscribe<A>(self: TxPubSub<A>): Effect<TxQueue<A, never>, never, Scope>

Example

(Subscribing multiple queues)

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

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<string>()

  return yield* Effect.scoped(
    Effect.gen(function*() {
      const sub1 = yield* TxPubSub.subscribe(hub)
      const sub2 = yield* TxPubSub.subscribe(hub)

      yield* TxPubSub.publish(hub, "broadcast")

      const msg1 = yield* TxQueue.take(sub1)
      const msg2 = yield* TxQueue.take(sub2)
      return [msg1, msg2]
    })
  )
})

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

Predicates

isEmpty

Added in v2.0.0 Source

Checks whether the TxPubSub has no pending messages (all subscriber queues are empty).

Signature

declare function isEmpty<A>(self: TxPubSub<A>): Effect<boolean>

Example

(Checking whether a pub/sub is empty)

import { Effect, TxPubSub } from "effect"

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<number>()
  return yield* TxPubSub.isEmpty(hub)
})

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

isFull

Added in v2.0.0 Source

Checks whether any subscriber queue is at capacity.

Signature

declare function isFull<A>(self: TxPubSub<A>): Effect<boolean>

Example

(Checking whether a pub/sub is full)

import { Effect, TxPubSub } from "effect"

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.bounded<number>(2)
  return yield* TxPubSub.isFull(hub)
})

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

isShutdown

Added in v2.0.0 Source

Checks whether the TxPubSub has been shut down.

Signature

declare function isShutdown<A>(self: TxPubSub<A>): Effect<boolean>

Example

(Checking whether a pub/sub is shut down)

import { Effect, TxPubSub } from "effect"

const program = Effect.gen(function*() {
  const hub = yield* TxPubSub.unbounded<number>()
  const before = yield* TxPubSub.isShutdown(hub)
  yield* TxPubSub.shutdown(hub)
  return [before, yield* TxPubSub.isShutdown(hub)]
})

await Effect.runPromise(program) // => [false, true]