ActorX for Gleam

ActorX - Reactive Extensions for Gleam using BEAM actors. A reactive programming library that composes BEAM actors for building asynchronous, event-driven applications.

This is a port of FSharp.Control.AsyncRx to Gleam, targeting the Erlang/BEAM runtime.

Installation

gleam add actorx

Example

import actorx
import gleam/io
import gleam/int

pub fn main() {
  // Create an observable pipeline
  let observable =
    actorx.from_list([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
    |> actorx.filter(fn(x) { x % 2 == 0 })  // Keep even numbers
    |> actorx.map(fn(x) { x * 10 })          // Multiply by 10
    |> actorx.take(3)                        // Take first 3

  // Create an observer
  let observer = actorx.make_observer(
    on_next: fn(x) { io.println("Value: " <> int.to_string(x)) },
    on_error: fn(_err) { Nil },
    on_completed: fn() { io.println("Done!") },
  )

  // Subscribe
  let _disposable = actorx.subscribe(observable, observer)
  // Output:
  // Value: 20
  // Value: 40
  // Value: 60
  // Done!
}

Builder Pattern with use

ActorX supports Gleam’s use keyword for monadic composition, similar to F#’s computation expressions:

import actorx
import actorx/builder.{bind, return}

pub fn example() {
  use x <- bind(actorx.single(10))
  use y <- bind(actorx.single(20))
  use z <- bind(actorx.from_list([1, 2, 3]))
  return(x + y + z)
}
// Emits: 31, 32, 33 then completes

Async Operators

ActorX provides time-based operators for async scenarios:

import actorx
import actorx/types.{Disposable}
import gleam/io
import gleam/int

pub fn async_example() {
  // Emit 0, 1, 2, ... every 100ms, take first 5
  let observable = actorx.interval(100)
    |> actorx.take(5)
    |> actorx.map(fn(x) { x * 10 })

  let observer = actorx.make_observer(
    on_next: fn(x) { io.println(int.to_string(x)) },
    on_error: fn(_) { Nil },
    on_completed: fn() { io.println("Done!") },
  )

  let Disposable(dispose) = actorx.subscribe(observable, observer)
  // Output over 500ms: 0, 10, 20, 30, 40, Done!

  // Can dispose early to cancel
  // dispose()
}

Subject Example

Subjects are both Observers and Observables - push values in, subscribe to receive them:

import actorx

pub fn subject_example() {
  let #(input, output) = actorx.single_subject()

  // Subscribe to output
  let _disp = actorx.subscribe(output, my_observer)

  // Push values through input
  actorx.on_next(input, 1)
  actorx.on_next(input, 2)
  actorx.on_completed(input)
}

Core Concepts

Observable

An Observable(a) represents a push-based stream of values of type a. Observables are lazy - they don’t produce values until subscribed to.

Observer

An Observer(a) receives notifications from an Observable:

• on_next(a) - Called for each value
• on_error(String) - Called on error (terminal)
• on_completed() - Called when complete (terminal)

The Rx contract guarantees: OnNext* (OnError | OnCompleted)?

Disposable

A Disposable represents a subscription that can be cancelled. Call dispose() to unsubscribe and release resources.

Available Operators

Creation

Transform

Filter

Timeshift (Async)

Subject

Combine

Error Handling

Builder (for use syntax)

Design

Why Gleam + BEAM?

The original F# AsyncRx uses MailboxProcessor (actors) to:

1. Serialize notifications (no concurrent observer calls)
2. Enforce Rx grammar
3. Manage state safely

Gleam on BEAM is a natural fit because:

• BEAM has native lightweight processes (actors)
• OTP provides battle-tested actor primitives
• Millions of concurrent subscriptions are feasible
• Supervision trees for fault tolerance

Architecture

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Observable │────▶│  Operator   │────▶│  Observer   │
│   (source)  │     │  (transform)│     │  (sink)     │
└─────────────┘     └─────────────┘     └─────────────┘
                           │
                    ┌──────┴──────┐
                    │ State Actor │
                    │ (for take,  │
                    │  skip, etc) │
                    └─────────────┘

Each stateful operator can use an actor to maintain state safely across async boundaries.

Current Implementation

ActorX uses two complementary approaches for state management:

Synchronous operators (like from_list, map, filter) use Erlang’s process dictionary for mutable state:

• Simple and performant for sync sources
• Avoids actor overhead for simple cases
• Works within a single process context

Asynchronous operators (like timer, interval, delay, debounce, throttle) use spawned processes:

• Each operator spawns a worker process that owns its state
• Communication via message passing with Gleam Subject
• Proper disposal via control messages to workers
• Safe across async boundaries

Compositional Design

Higher-order operators are composed from primitives, following the standard Rx pattern:

// flat_map = map + merge_inner
pub fn flat_map(source, mapper) {
  source |> map(mapper) |> merge_inner()
}

// concat_map = map + concat_inner
pub fn concat_map(source, mapper) {
  source |> map(mapper) |> concat_inner()
}

This enables building new operators (like switch_map = map + switch_inner) from reusable primitives.

Safe Observer

The safe_observer module provides Rx grammar enforcement:

• Tracks “stopped” state
• Ignores events after terminal
• Calls disposal on terminal events

Roadmap

Phase 1: Core Operators ✓

• Basic types (Observable, Observer, Disposable, Notification)
• Creation operators (single, empty, never, fail, from_list, defer)
• Transform operators (map, flat_map, concat_map)
• Filter operators (filter, take, skip, take_while, distinct_until_changed, etc.)
• Safe observer with Rx grammar enforcement
• Builder module with use keyword support
• Test suite

Phase 2: Actor-Based Async ✓

• timer operator - emit after delay
• interval operator - emit at regular intervals
• delay operator - delay emissions
• debounce operator - emit after silence
• throttle operator - rate limiting
• single_subject - single-subscriber subject
• Async disposal support

Phase 3: Combining Operators ✓

• merge - Merge multiple observables
• combine_latest - Combine latest values
• with_latest_from - Sample with latest
• zip - Pair elements by index

Phase 4: Advanced Features ✓

• subject - Multicast hot observable
• scan / reduce - Aggregation
• retry / catch - Error handling
• share / publish - Share subscriptions
• group_by - Grouped streams

Phase 5: Integration

• Supervision integration
• Telemetry/metrics

Examples

Timeflies Demo

A classic Rx demo where letters trail behind your mouse cursor. Demonstrates subject, flat_map, delay, and WebSocket integration with Mist.

cd examples/timeflies
gleam run
# Open http://localhost:3000

Development

gleam build  # Build the project
gleam test   # Run the tests

License

MIT

Related Projects

• FSharp.Control.AsyncRx - Original F# implementation
• RxPY - ReactiveX for Python
• Reaxive - ReactiveX for Elixir
• GenStage - Elixir’s demand-driven streams