<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)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.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: SubscriptionRef<A>): Stream.Stream<A>A> => import StreamStream.const fromPubSub: <A>(
pubsub: PubSub.PubSub<A>
) => Stream<A>
Creates a stream from a subscription to a PubSub.
Example (Creating a stream from a subscription to a PubSub)
import { Console, Effect, Fiber, PubSub, Stream } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.unbounded<number>()
const fiber = yield* Stream.fromPubSub(pubsub).pipe(
Stream.take(3),
Stream.runCollect,
Effect.forkChild
)
yield* PubSub.publish(pubsub, 1)
yield* PubSub.publish(pubsub, 2)
yield* PubSub.publish(pubsub, 3)
const values = yield* Fiber.join(fiber)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3 ]
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<A>.pubsub: PubSub.PubSub<A>(property) SubscriptionRef<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; <…;
}