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.

19 exports Since v2.0.0 Source

Constructors

bounded

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v4.0.0

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

Since v4.0.0

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

unstable

Since v4.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

unstable

Since v4.0.0

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

Since v2.0.0

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]

unstable

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

isNonEmpty

Since v4.0.0

Checks whether any subscriber queue has pending messages. A pub/sub with no subscribers is empty, even if values have been published.

Signature

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

Example Checking whether a pub/sub is non-empty

import { Effect, TxPubSub } from "effect"

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

  const nonEmpty = yield* Effect.scoped(Effect.gen(function*() {
    yield* TxPubSub.subscribe(hub)
    yield* TxPubSub.publish(hub, 1)
    return yield* TxPubSub.isNonEmpty(hub)
  }))
  return [empty, nonEmpty] as const
})

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

isShutdown

Since v2.0.0

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]