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)

1
import { SubscriptionRef } from "effect"
2

3
const ref = SubscriptionRef.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:

1
import { Ref, Effect } from "effect"
2

3
// Server function that increments a shared value forever
4
const server = (ref: Ref.Ref<number>) => Ref.update(ref, (n) => n + 1).pipe(Effect.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:

1
import { Ref, Effect, Stream, Random } from "effect"
2

3
// Server function that increments a shared value forever
4
const server = (ref: Ref.Ref<number>) => Ref.update(ref, (n) => n + 1).pipe(Effect.forever)
5

6
// Client function that observes the stream of changes
7
const client = (changes: Stream.Stream<number>) =>
8
  Effect.gen(function* () {
9
    const n = yield* Random.nextIntBetween(1, 10)
10
    const chunk = yield* Stream.runCollect(Stream.take(changes, n))
11
    return chunk
12
  })

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.

1
import { Ref, Effect, Stream, Random, SubscriptionRef, Fiber } from "effect"
2

3
// Server function that increments a shared value forever
4
const server = (ref: Ref.Ref<number>) => Ref.update(ref, (n) => n + 1).pipe(Effect.forever)
5

6
// Client function that observes the stream of changes
7
const client = (changes: Stream.Stream<number>) =>
8
  Effect.gen(function* () {
9
    const n = yield* Random.nextIntBetween(1, 10)
10
    const chunk = yield* Stream.runCollect(Stream.take(changes, n))
11
    return chunk
12
  })
13

14
const program = Effect.gen(function* () {
15
  // Create a SubscriptionRef with an initial value of 0
16
  const ref = yield* SubscriptionRef.make(0)
17

18
  // Fork the server to run concurrently
19
  const serverFiber = yield* Effect.fork(server(ref))
20

21
  // Create 5 clients that subscribe to the changes stream
22
  const clients = new Array(5).fill(null).map(() => client(ref.changes))
23

24
  // Run all clients in concurrently and collect their results
25
  const chunks = yield* Effect.all(clients, { concurrency: "unbounded" })
26

27
  // Interrupt the server when clients are done
28
  yield* Fiber.interrupt(serverFiber)
29

30
  // Output the results collected by each client
31
  for (const chunk of chunks) {
32
    console.log(chunk)
33
  }
34
})
35

36
Effect.runPromise(program)
37
/*
38
Example Output:
39
{ _id: 'Chunk', values: [ 4, 5, 6, 7, 8, 9 ] }
40
{ _id: 'Chunk', values: [ 4 ] }
41
{ _id: 'Chunk', values: [ 4, 5, 6, 7, 8, 9 ] }
42
{ _id: 'Chunk', values: [ 4, 5 ] }
43
{ _id: 'Chunk', values: [ 4, 5, 6, 7, 8, 9 ] }
44
*/

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.