<A>(source: SubscriptionRef.SubscriptionRef<A>): Subscribable<A>Build a Subscribable view over a SubscriptionRef — the impl side of a ref field: the
impl owns the ref (writes it), consumers get read + observe.
export const const subscribable: <A>(
source: SubscriptionRef.SubscriptionRef<A>
) => Subscribable<A>
Build a
Subscribable
view over a SubscriptionRef — the impl side of a
ref
field: the
impl owns the ref (writes it), consumers get read + observe.
subscribable = <function (type parameter) A in <A>(source: SubscriptionRef.SubscriptionRef<A>): Subscribable<A>A>(
source: SubscriptionRef.SubscriptionRef<A>(parameter) source: {
value: A;
semaphore: Semaphore.Semaphore;
pubsub: PubSub.PubSub<A>;
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; <…;
}
import SubscriptionRefSubscriptionRef.interface SubscriptionRef<in out A>A mutable reference whose updates are serialized and published to
subscribers.
When to use
Use to observe the current value and subsequent updates as a
stream.
The SubscriptionRef namespace containing type definitions associated with
subscription references.
SubscriptionRef<function (type parameter) A in <A>(source: SubscriptionRef.SubscriptionRef<A>): Subscribable<A>A>,
): interface Subscribable<A>A read-only reactive value: its current value (
Subscribable.get
, an Effect) plus a stream
of every change (
Subscribable.changes
). This is what a
ref
field surfaces — uniform local
and remote — and it's exactly the read side of a SubscriptionRef (Effect ships no Subscribable type in
this beta, so we name it here).
Subscribable<function (type parameter) A in <A>(source: SubscriptionRef.SubscriptionRef<A>): Subscribable<A>A> => ({
Subscribable<A>.get: Effect.Effect<A, never, never>(property) Subscribable<A>.get: {
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; <…;
toString: () => string;
toJSON: () => unknown;
}
import SubscriptionRefSubscriptionRef.const get: <A>(
self: SubscriptionRef<A>
) => Effect.Effect<A>
Retrieves the current value of the SubscriptionRef.
Example (Reading the current value)
import { Effect, SubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* SubscriptionRef.make(42)
const value = yield* SubscriptionRef.get(ref)
console.log(value)
})
get(source: SubscriptionRef.SubscriptionRef<A>(parameter) source: {
value: A;
semaphore: Semaphore.Semaphore;
pubsub: PubSub.PubSub<A>;
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; <…;
}
Subscribable<A>.changes: Stream.Stream<A, never, never>(property) Subscribable<A>.changes: {
channel: Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E, void, unknown, unknown, unknown, R>;
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; <…;
}
import SubscriptionRefSubscriptionRef.const changes: <A>(
self: SubscriptionRef<A>
) => Stream.Stream<A>
Creates a stream that emits the current value and all subsequent changes to
the SubscriptionRef.
Details
The stream will first emit the current value, then emit all future changes
as they occur.
Example (Streaming changes)
import { Deferred, Effect, Fiber, Stream, SubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* SubscriptionRef.make(0)
const ready = yield* Deferred.make<void>()
const fiber = yield* SubscriptionRef.changes(ref).pipe(
Stream.tap(() => Deferred.succeed(ready, void 0)),
Stream.take(3),
Stream.runCollect,
Effect.forkChild
)
yield* Deferred.await(ready)
yield* SubscriptionRef.set(ref, 1)
yield* SubscriptionRef.set(ref, 2)
const values = yield* Fiber.join(fiber)
console.log(values) // [ 0, 1, 2 ]
})
Effect.runPromise(program)
changes(source: SubscriptionRef.SubscriptionRef<A>(parameter) source: {
value: A;
semaphore: Semaphore.Semaphore;
pubsub: PubSub.PubSub<A>;
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; <…;
}