<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.type Stream.Stream = /*unresolved*/ anyStream<function (type parameter) A in <A>(self: TxSubscriptionRef<A>): Stream.Stream<A, never, never>A, never, never> =>
import StreamStream.unwrap(
import EffectEffect.map(
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.fromEffectRepeat(import EffectEffect.tx(import TxQueueTxQueue.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;
}