<A>(self: TxSubscriptionRef<A>): Stream.Stream<A, never, never>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.
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))
)
console.log(values) // [2]
})export const const changesStream: <A>(
self: TxSubscriptionRef<A>
) => Stream.Stream<A, never, never>
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.
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))
)
console.log(values) // [2]
})
changesStream = <function (type parameter) A in <A>(self: TxSubscriptionRef<A>): Stream.Stream<A, never, never>A>(self: TxSubscriptionRef<A>(parameter) self: {
ref: TxRef.TxRef<A>;
pubsub: TxPubSub.TxPubSub<A>;
toString: () => string;
toJSON: () => unknown;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface TxSubscriptionRef<in out A>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.
Example (Subscribing to transactional changes)
import { Effect, TxQueue, TxSubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* TxSubscriptionRef.make(0)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxSubscriptionRef.changes(ref)
const initial = yield* TxQueue.take(sub)
console.log(initial) // 0
yield* TxSubscriptionRef.set(ref, 1)
const next = yield* TxQueue.take(sub)
console.log(next) // 1
})
)
})
TxSubscriptionRef<function (type parameter) A in <A>(self: TxSubscriptionRef<A>): Stream.Stream<A, never, never>A>): import StreamStream.interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A>(self: TxSubscriptionRef<A>): Stream.Stream<A, never, never>A, never, never> =>
import StreamStream.const unwrap: <A, E2, R2, E, R>(
effect: Effect.Effect<Stream<A, E2, R2>, E, R>
) => Stream<
A,
E | E2,
R2 | Exclude<R, Scope.Scope>
>
Creates a stream produced from an Effect.
Example (Unwrapping a stream effect)
import { Console, Effect, Stream } from "effect"
const effect = Effect.succeed(Stream.make(1, 2, 3))
const stream = Stream.unwrap(effect)
const program = Effect.gen(function*() {
const chunk = yield* Stream.runCollect(stream)
yield* Console.log(chunk)
})
// [1, 2, 3]
unwrap(
import EffectEffect.const map: {
<A, B>(f: (a: A) => B): <E, R>(
self: Effect<A, E, R>
) => Effect<B, E, R>
<A, E, R, B>(
self: Effect<A, E, R>,
f: (a: A) => B
): Effect<B, E, R>
}
const changes: <A>(
self: TxSubscriptionRef<A>
) => Effect.Effect<
TxQueue.TxQueue<A>,
never,
Scope.Scope
>
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.
Example (Subscribing to changes)
import { Effect, TxQueue, TxSubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* TxSubscriptionRef.make(0)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxSubscriptionRef.changes(ref)
const initial = yield* TxQueue.take(sub)
console.log(initial) // 0
yield* TxSubscriptionRef.set(ref, 1)
const next = yield* TxQueue.take(sub)
console.log(next) // 1
})
)
})
changes(self: TxSubscriptionRef<A>(parameter) self: {
ref: TxRef.TxRef<A>;
pubsub: TxPubSub.TxPubSub<A>;
toString: () => string;
toJSON: () => unknown;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
sub: TxQueue.TxQueue<A, never>(parameter) sub: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
import StreamStream.const fromEffectRepeat: <A, E, R>(
effect: Effect.Effect<A, E, R>
) => Stream<A, Pull.ExcludeDone<E>, R>
Creates a stream from an effect producing a value of type A which repeats forever.
Example (Repeating an effect forever)
import { Console, Effect, Random, Stream } from "effect"
const program = Effect.gen(function*() {
const stream = Stream.fromEffectRepeat(Random.nextInt).pipe(
Stream.take(5)
)
const values = yield* Stream.runCollect(stream)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 3891571149, 4239494205, 2352981603, 2339111046, 1488052210 ]
fromEffectRepeat(import EffectEffect.const tx: <A, E, R>(
effect: Effect<A, E, R>
) => Effect<A, E, Exclude<R, Transaction>>
Defines a transaction boundary. Transactions are "all or nothing" with respect to changes
made to transactional values (i.e. TxRef) that occur within the transaction body.
Details
If called inside an active transaction, tx composes with the current transaction and reuses
its journal and retry state instead of creating a nested boundary.
Effect transactions are optimistic with retry. A transaction is retried when
its body explicitly calls Effect.txRetry and any accessed transactional
value changes, or when any accessed transactional value changes because a
different transaction commits before the current one.
The outermost tx call creates the transaction boundary and commits or rolls back the full
composed transaction.
Example (Running a transaction)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
const ref1 = yield* TxRef.make(0)
const ref2 = yield* TxRef.make(0)
// Nested tx calls compose into the same transaction
yield* Effect.tx(Effect.gen(function*() {
yield* TxRef.set(ref1, 10)
yield* Effect.tx(TxRef.set(ref2, 20))
const sum = (yield* TxRef.get(ref1)) + (yield* TxRef.get(ref2))
console.log(`Transaction sum: ${sum}`)
}))
console.log(`Final ref1: ${yield* TxRef.get(ref1)}`) // 10
console.log(`Final ref2: ${yield* TxRef.get(ref2)}`) // 20
})
tx(import TxQueueTxQueue.const take: <A, E>(
self: TxDequeue<A, E>
) => Effect.Effect<A, E>
Takes the next item from the queue, retrying the transaction while the queue
is empty.
Details
If the queue is done, the effect fails with the queue's completion cause. This function mutates the original TxQueue by removing the first item. It does not return a new TxQueue reference.
Example (Taking a value)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offer(queue, 42)
// Take an item - blocks if empty
const item = yield* TxQueue.take(queue)
console.log(item) // 42
// When queue fails, take fails with the same error
yield* TxQueue.fail(queue, "queue error")
const result = yield* Effect.flip(TxQueue.take(queue))
console.log(result) // "queue error"
})
take(sub: TxQueue.TxQueue<A, never>(parameter) sub: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}