Skip to content

SubscriptionRef

A SubscriptionRef<A> is a specialized form of a SynchronizedRef. It allows us to subscribe and receive updates on the current value and any changes made to that value.

interface SubscriptionRef<A> extends SynchronizedRef<A> {
/**
* A stream containing the current value of the `Ref` as well as all changes
* to that value.
*/
readonly changes: Stream<A>
}

You can perform all standard operations on a SubscriptionRef, such as get, set, or modify to interact with the current value.

The key feature of SubscriptionRef is its changes stream. This stream allows you to observe the current value at the moment of subscription and receive all subsequent changes. Every time the stream is run, it emits the current value and tracks future updates.

To create a SubscriptionRef, you can use the SubscriptionRef.make constructor, specifying the initial value:

Example (Creating a SubscriptionRef)

import {
import SubscriptionRef
SubscriptionRef
} from "effect"
const
const ref: Effect<SubscriptionRef.SubscriptionRef<number>, never, never>
ref
=
import SubscriptionRef
SubscriptionRef
.
const make: <number>(value: number) => Effect<SubscriptionRef.SubscriptionRef<number>, never, never>

Creates a new SubscriptionRef with the specified value.

@since2.0.0

make
(0)

SubscriptionRef is particularly useful for modeling shared state when multiple observers need to react to changes. For example, in functional reactive programming, the SubscriptionRef could represent a portion of the application state, and various observers (like UI components) would update in response to state changes.

Example (Server-Client Model with SubscriptionRef)

In the following example, a “server” continually updates a shared value, while multiple “clients” observe the changes:

import {
import Ref
Ref
,
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
} from "effect"
// Server function that increments a shared value forever
const
const server: (ref: Ref.Ref<number>) => Effect.Effect<never, never, never>
server
= (
ref: Ref.Ref<number>
ref
:
import Ref
Ref
.
interface Ref<in out A>

@since2.0.0

@since2.0.0

Ref
<number>) =>
import Ref
Ref
.
const update: <number>(self: Ref.Ref<number>, f: (a: number) => number) => Effect.Effect<void> (+1 overload)

@since2.0.0

update
(
ref: Ref.Ref<number>
ref
, (
n: number
n
) =>
n: number
n
+ 1).
Pipeable.pipe<Effect.Effect<void, never, never>, Effect.Effect<never, never, never>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<void, never, never>) => Effect.Effect<never, never, never>): Effect.Effect<...> (+21 overloads)
pipe
(
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const forever: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<never, E, R>

Repeats this effect forever (until the first error).

@since2.0.0

forever
)

The server function operates on a regular Ref and continuously updates the value. It doesn’t need to know about SubscriptionRef directly.

Next, let’s define a client that subscribes to changes and collects a specified number of values:

import {
import Ref
Ref
,
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
,
import Stream
Stream
,
import Random
Random
} from "effect"
// Server function that increments a shared value forever
const
const server: (ref: Ref.Ref<number>) => Effect.Effect<never, never, never>
server
= (
ref: Ref.Ref<number>
ref
:
import Ref
Ref
.
interface Ref<in out A>

@since2.0.0

@since2.0.0

Ref
<number>) =>
import Ref
Ref
.
const update: <number>(self: Ref.Ref<number>, f: (a: number) => number) => Effect.Effect<void> (+1 overload)

@since2.0.0

update
(
ref: Ref.Ref<number>
ref
, (
n: number
n
) =>
n: number
n
+ 1).
Pipeable.pipe<Effect.Effect<void, never, never>, Effect.Effect<never, never, never>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<void, never, never>) => Effect.Effect<never, never, never>): Effect.Effect<...> (+21 overloads)
pipe
(
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const forever: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<never, E, R>

Repeats this effect forever (until the first error).

@since2.0.0

forever
)
// Client function that observes the stream of changes
const
const client: (changes: Stream.Stream<number>) => Effect.Effect<Chunk<number>, never, never>
client
= (
changes: Stream.Stream<number, never, never>
changes
:
import Stream
Stream
.
interface Stream<out A, out E = never, out R = never>

A Stream<A, E, R> is a description of a program that, when evaluated, may emit zero or more values of type A, may fail with errors of type E, and uses an context of type R. One way to think of Stream is as a Effect program that could emit multiple values.

Stream is a purely functional pull based stream. Pull based streams offer inherent laziness and backpressure, relieving users of the need to manage buffers between operators. As an optimization, Stream does not emit single values, but rather an array of values. This allows the cost of effect evaluation to be amortized.

Stream forms a monad on its A type parameter, and has error management facilities for its E type parameter, modeled similarly to Effect (with some adjustments for the multiple-valued nature of Stream). These aspects allow for rich and expressive composition of streams.

@since2.0.0

@since2.0.0

Stream
<number>) =>
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const gen: <YieldWrap<Effect.Effect<number, never, never>> | YieldWrap<Effect.Effect<Chunk<number>, never, never>>, Chunk<number>>(f: (resume: Effect.Adapter) => Generator<...>) => Effect.Effect<...> (+1 overload)

@since2.0.0

gen
(function* () {
const
const n: number
n
= yield*
import Random
Random
.
const nextIntBetween: (min: number, max: number) => Effect.Effect<number>

Returns the next integer value in the specified range from the pseudo-random number generator.

@since2.0.0

nextIntBetween
(1, 10)
const
const chunk: Chunk<number>
chunk
= yield*
import Stream
Stream
.
const runCollect: <number, never, never>(self: Stream.Stream<number, never, never>) => Effect.Effect<Chunk<number>, never, never>

Runs the stream and collects all of its elements to a chunk.

@since2.0.0

runCollect
(
import Stream
Stream
.
const take: <number, never, never>(self: Stream.Stream<number, never, never>, n: number) => Stream.Stream<number, never, never> (+1 overload)

Takes the specified number of elements from this stream.

@example

import { Effect, Stream } from "effect"

const stream = Stream.take(Stream.iterate(0, (n) => n + 1), 5)

// Effect.runPromise(Stream.runCollect(stream)).then(console.log) // { _id: 'Chunk', values: [ 0, 1, 2, 3, 4 ] }

@since2.0.0

take
(
changes: Stream.Stream<number, never, never>
changes
,
const n: number
n
))
return
const chunk: Chunk<number>
chunk
})

Similarly, the client function only works with a Stream of values and doesn’t concern itself with the source of these values.

To tie everything together, we start the server, launch multiple client instances in parallel, and then shut down the server when we’re finished. We also create the SubscriptionRef in this process.

import {
import Ref
Ref
,
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
,
import Stream
Stream
,
import Random
Random
,
import SubscriptionRef
SubscriptionRef
,
import Fiber
Fiber
} from "effect"
// Server function that increments a shared value forever
const
const server: (ref: Ref.Ref<number>) => Effect.Effect<never, never, never>
server
= (
ref: Ref.Ref<number>
ref
:
import Ref
Ref
.
interface Ref<in out A>

@since2.0.0

@since2.0.0

Ref
<number>) =>
import Ref
Ref
.
const update: <number>(self: Ref.Ref<number>, f: (a: number) => number) => Effect.Effect<void> (+1 overload)

@since2.0.0

update
(
ref: Ref.Ref<number>
ref
, (
n: number
n
) =>
n: number
n
+ 1).
Pipeable.pipe<Effect.Effect<void, never, never>, Effect.Effect<never, never, never>>(this: Effect.Effect<...>, ab: (_: Effect.Effect<void, never, never>) => Effect.Effect<never, never, never>): Effect.Effect<...> (+21 overloads)
pipe
(
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const forever: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<never, E, R>

Repeats this effect forever (until the first error).

@since2.0.0

forever
)
// Client function that observes the stream of changes
const
const client: (changes: Stream.Stream<number>) => Effect.Effect<Chunk<number>, never, never>
client
= (
changes: Stream.Stream<number, never, never>
changes
:
import Stream
Stream
.
interface Stream<out A, out E = never, out R = never>

A Stream<A, E, R> is a description of a program that, when evaluated, may emit zero or more values of type A, may fail with errors of type E, and uses an context of type R. One way to think of Stream is as a Effect program that could emit multiple values.

Stream is a purely functional pull based stream. Pull based streams offer inherent laziness and backpressure, relieving users of the need to manage buffers between operators. As an optimization, Stream does not emit single values, but rather an array of values. This allows the cost of effect evaluation to be amortized.

Stream forms a monad on its A type parameter, and has error management facilities for its E type parameter, modeled similarly to Effect (with some adjustments for the multiple-valued nature of Stream). These aspects allow for rich and expressive composition of streams.

@since2.0.0

@since2.0.0

Stream
<number>) =>
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const gen: <YieldWrap<Effect.Effect<number, never, never>> | YieldWrap<Effect.Effect<Chunk<number>, never, never>>, Chunk<number>>(f: (resume: Effect.Adapter) => Generator<...>) => Effect.Effect<...> (+1 overload)

@since2.0.0

gen
(function* () {
const
const n: number
n
= yield*
import Random
Random
.
const nextIntBetween: (min: number, max: number) => Effect.Effect<number>

Returns the next integer value in the specified range from the pseudo-random number generator.

@since2.0.0

nextIntBetween
(1, 10)
const
const chunk: Chunk<number>
chunk
= yield*
import Stream
Stream
.
const runCollect: <number, never, never>(self: Stream.Stream<number, never, never>) => Effect.Effect<Chunk<number>, never, never>

Runs the stream and collects all of its elements to a chunk.

@since2.0.0

runCollect
(
import Stream
Stream
.
const take: <number, never, never>(self: Stream.Stream<number, never, never>, n: number) => Stream.Stream<number, never, never> (+1 overload)

Takes the specified number of elements from this stream.

@example

import { Effect, Stream } from "effect"

const stream = Stream.take(Stream.iterate(0, (n) => n + 1), 5)

// Effect.runPromise(Stream.runCollect(stream)).then(console.log) // { _id: 'Chunk', values: [ 0, 1, 2, 3, 4 ] }

@since2.0.0

take
(
changes: Stream.Stream<number, never, never>
changes
,
const n: number
n
))
return
const chunk: Chunk<number>
chunk
})
const
const program: Effect.Effect<void, never, never>
program
=
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const gen: <YieldWrap<Effect.Effect<SubscriptionRef.SubscriptionRef<number>, never, never>> | YieldWrap<Effect.Effect<Fiber.RuntimeFiber<never, never>, never, never>> | YieldWrap<...> | YieldWrap<...>, void>(f: (resume: Effect.Adapter) => Generator<...>) => Effect.Effect<...> (+1 overload)

@since2.0.0

gen
(function* () {
// Create a SubscriptionRef with an initial value of 0
const
const ref: SubscriptionRef.SubscriptionRef<number>
ref
= yield*
import SubscriptionRef
SubscriptionRef
.
const make: <number>(value: number) => Effect.Effect<SubscriptionRef.SubscriptionRef<number>, never, never>

Creates a new SubscriptionRef with the specified value.

@since2.0.0

make
(0)
// Fork the server to run concurrently
const
const serverFiber: Fiber.RuntimeFiber<never, never>
serverFiber
= yield*
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const fork: <never, never, never>(self: Effect.Effect<never, never, never>) => Effect.Effect<Fiber.RuntimeFiber<never, never>, never, never>

Returns an effect that forks this effect into its own separate fiber, returning the fiber immediately, without waiting for it to begin executing the effect.

You can use the fork method whenever you want to execute an effect in a new fiber, concurrently and without "blocking" the fiber executing other effects. Using fibers can be tricky, so instead of using this method directly, consider other higher-level methods, such as raceWith, zipPar, and so forth.

The fiber returned by this method has methods to interrupt the fiber and to wait for it to finish executing the effect. See Fiber for more information.

Whenever you use this method to launch a new fiber, the new fiber is attached to the parent fiber's scope. This means when the parent fiber terminates, the child fiber will be terminated as well, ensuring that no fibers leak. This behavior is called "auto supervision", and if this behavior is not desired, you may use the forkDaemon or forkIn methods.

@since2.0.0

fork
(
const server: (ref: Ref.Ref<number>) => Effect.Effect<never, never, never>
server
(
const ref: SubscriptionRef.SubscriptionRef<number>
ref
))
// Create 5 clients that subscribe to the changes stream
const
const clients: Effect.Effect<Chunk<number>, never, never>[]
clients
= new
var Array: ArrayConstructor
new (arrayLength?: number) => any[] (+2 overloads)
Array
(5).
Array<any>.fill(value: any, start?: number, end?: number): any[]

Changes all array elements from start to end index to a static value and returns the modified array

@paramvalue value to fill array section with

@paramstart index to start filling the array at. If start is negative, it is treated as length+start where length is the length of the array.

@paramend index to stop filling the array at. If end is negative, it is treated as length+end.

fill
(null).
Array<any>.map<Effect.Effect<Chunk<number>, never, never>>(callbackfn: (value: any, index: number, array: any[]) => Effect.Effect<Chunk<number>, never, never>, thisArg?: any): Effect.Effect<...>[]

Calls a defined callback function on each element of an array, and returns an array that contains the results.

@paramcallbackfn A function that accepts up to three arguments. The map method calls the callbackfn function one time for each element in the array.

@paramthisArg An object to which the this keyword can refer in the callbackfn function. If thisArg is omitted, undefined is used as the this value.

map
(() =>
const client: (changes: Stream.Stream<number>) => Effect.Effect<Chunk<number>, never, never>
client
(
const ref: SubscriptionRef.SubscriptionRef<number>
ref
.
SubscriptionRef<number>.changes: Stream.Stream<number, never, never>

A stream containing the current value of the Ref as well as all changes to that value.

changes
))
// Run all clients in concurrently and collect their results
const
const chunks: Chunk<number>[]
chunks
= yield*
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const all: <Effect.Effect<Chunk<number>, never, never>[], {
concurrency: "unbounded";
}>(arg: Effect.Effect<Chunk<number>, never, never>[], options?: {
concurrency: "unbounded";
} | undefined) => Effect.Effect<...>

Runs all the provided effects in sequence respecting the structure provided in input.

Supports multiple arguments, a single argument tuple / array or record / struct.

@since2.0.0

all
(
const clients: Effect.Effect<Chunk<number>, never, never>[]
clients
, {
concurrency: "unbounded"
concurrency
: "unbounded" })
// Interrupt the server when clients are done
yield*
import Fiber
Fiber
.
const interrupt: <never, never>(self: Fiber.Fiber<never, never>) => Effect.Effect<Exit<never, never>, never, never>

Interrupts the fiber from whichever fiber is calling this method. If the fiber has already exited, the returned effect will resume immediately. Otherwise, the effect will resume when the fiber exits.

@since2.0.0

interrupt
(
const serverFiber: Fiber.RuntimeFiber<never, never>
serverFiber
)
// Output the results collected by each client
for (const
const chunk: Chunk<number>
chunk
of
const chunks: Chunk<number>[]
chunks
) {
var console: Console

The console module provides a simple debugging console that is similar to the JavaScript console mechanism provided by web browsers.

The module exports two specific components:

  • A Console class with methods such as console.log(), console.error() and console.warn() that can be used to write to any Node.js stream.
  • A global console instance configured to write to process.stdout and process.stderr. The global console can be used without importing the node:console module.

Warning: The global console object's methods are neither consistently synchronous like the browser APIs they resemble, nor are they consistently asynchronous like all other Node.js streams. See the note on process I/O for more information.

Example using the global console:

console.log('hello world');
// Prints: hello world, to stdout
console.log('hello %s', 'world');
// Prints: hello world, to stdout
console.error(new Error('Whoops, something bad happened'));
// Prints error message and stack trace to stderr:
// Error: Whoops, something bad happened
// at [eval]:5:15
// at Script.runInThisContext (node:vm:132:18)
// at Object.runInThisContext (node:vm:309:38)
// at node:internal/process/execution:77:19
// at [eval]-wrapper:6:22
// at evalScript (node:internal/process/execution:76:60)
// at node:internal/main/eval_string:23:3
const name = 'Will Robinson';
console.warn(`Danger ${name}! Danger!`);
// Prints: Danger Will Robinson! Danger!, to stderr

Example using the Console class:

const out = getStreamSomehow();
const err = getStreamSomehow();
const myConsole = new console.Console(out, err);
myConsole.log('hello world');
// Prints: hello world, to out
myConsole.log('hello %s', 'world');
// Prints: hello world, to out
myConsole.error(new Error('Whoops, something bad happened'));
// Prints: [Error: Whoops, something bad happened], to err
const name = 'Will Robinson';
myConsole.warn(`Danger ${name}! Danger!`);
// Prints: Danger Will Robinson! Danger!, to err

@seesource

console
.
Console.log(message?: any, ...optionalParams: any[]): void

Prints to stdout with newline. Multiple arguments can be passed, with the first used as the primary message and all additional used as substitution values similar to printf(3) (the arguments are all passed to util.format()).

const count = 5;
console.log('count: %d', count);
// Prints: count: 5, to stdout
console.log('count:', count);
// Prints: count: 5, to stdout

See util.format() for more information.

@sincev0.1.100

log
(
const chunk: Chunk<number>
chunk
)
}
})
import Effect

@since2.0.0

@since2.0.0

@since2.0.0

Effect
.
const runPromise: <void, never>(effect: Effect.Effect<void, never, never>, options?: {
readonly signal?: AbortSignal;
} | undefined) => Promise<void>

Executes an effect and returns a Promise that resolves with the result.

Use runPromise when working with asynchronous effects and you need to integrate with code that uses Promises. If the effect fails, the returned Promise will be rejected with the error.

@example

import { Effect } from "effect"

// Execute an effect and handle the result with a Promise Effect.runPromise(Effect.succeed(1)).then(console.log) // Output: 1

// Execute a failing effect and handle the rejection Effect.runPromise(Effect.fail("my error")).catch((error) => { console.error("Effect failed with error:", error) })

@since2.0.0

runPromise
(
const program: Effect.Effect<void, never, never>
program
)
/*
Example Output:
{ _id: 'Chunk', values: [ 4, 5, 6, 7, 8, 9 ] }
{ _id: 'Chunk', values: [ 4 ] }
{ _id: 'Chunk', values: [ 4, 5, 6, 7, 8, 9 ] }
{ _id: 'Chunk', values: [ 4, 5 ] }
{ _id: 'Chunk', values: [ 4, 5, 6, 7, 8, 9 ] }
*/

This setup ensures that each client observes the current value when it starts and receives all subsequent changes to the value.

Since the changes are represented as streams, you can easily build more complex programs using familiar stream operators. You can transform, filter, or merge these streams with other streams to achieve more sophisticated behavior.