<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 effect/SubscriptionRef.ts:1601 lines
export const 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 = <function (type parameter) A in <A>(self: SubscriptionRef<A>): Stream.Stream<A>A>(self: SubscriptionRef<A>(parameter) self: {
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; <…;
}
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>(self: SubscriptionRef<A>): Stream.Stream<A>A>): import StreamStream.type Stream.Stream = /*unresolved*/ anyStream<function (type parameter) A in <A>(self: SubscriptionRef<A>): Stream.Stream<A>A> => import StreamStream.fromPubSub(self: SubscriptionRef<A>(parameter) self: {
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; <…;
}
SubscriptionRef<in out A>.pubsub: PubSub.PubSub<A>(property) SubscriptionRef<in out A>.pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<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; <…;
}
Referenced by 2 symbols