Returns the observable that emits first. Also known as race. Other sources are ignored after one emits.

Creates a cold Observable that performs a request-response call to an actor.

Each subscription triggers a new request. The make_request function receives a reply Subject and should return the message to send to the actor.

Example

type CounterMsg {
  Increment(Int)
  GetValue(Subject(Int))
}

// Single call
actorx.call_actor(counter, 1000, GetValue)
|> actorx.subscribe(observer)

// Periodic polling
actorx.interval(1000)
|> actorx.flat_map(fn(_) {
  actorx.call_actor(counter, 1000, GetValue)
})
|> actorx.subscribe(value_observer)

On error, switches to a fallback observable returned by the handler. Also known as catch_error or on_error_resume_next.

Example

// On error, emit a default value
risky_observable
|> catch(fn(_error) { single(default_value) })

Filter and map in one operation.

Combines the latest values from two observable sequences. Emits whenever either source emits, after both have emitted at least once. Completes when both sources complete.

Concatenates multiple observables sequentially. Subscribes to each source only after the previous completes.

Concatenates two observables sequentially.

Flattens an Observable of Observables by concatenating in order.

Subscribes to each inner observable only after the previous one completes. Queues inner observables and processes them sequentially.

Project each element to an observable and concatenate in order.

Composed from map and concat_inner: concat_map(source, f) = source |> map(f) |> concat_inner()

Unlike flat_map, this preserves the order of inner observables.

Project each element and its index to an observable and concatenate in order.

Composed from mapi and concat_inner: concat_mapi(source, f) = source |> mapi(f) |> concat_inner()

Create an observable from a subscribe function.

Emits a value only after the specified time has passed without another value being emitted.

Emit default value if source completes empty.

Create an observable that calls a factory function on each subscription.

Delays each emission from the source observable by the specified time.

Filter out all duplicate values (not just consecutive).

Emit only when value changes from previous.

Create an observable that completes immediately without emitting.

Create an empty disposable.

Create an observable that errors immediately.

Filter elements based on a predicate.

Take only the first element. Errors if source is empty.

Project each element to an observable and flatten.

Composed from map and merge_inner: flat_map(source, f) = source |> map(f) |> merge_inner()

Project each element and its index to an observable and flatten.

Composed from mapi and merge_inner: flat_mapi(source, f) = source |> mapi(f) |> merge_inner()

Waits for all observables to complete, then emits list of their last values.

Create an observable from a list of values.

Creates a Subject and Observable pair for bridging actors.

Returns a tuple of (Subject, Observable) where:

• Values sent to the Subject are emitted to all Observable subscribers
• The Observable never completes on its own - use take_until for completion

This is useful for having an actor publish events that can be consumed reactively via the Observable.

Example

let #(events_subject, events_observable) = actorx.from_subject()

// Pass subject to a producer actor
let _producer = start_event_producer(events_subject)

// Subscribe to the observable
events_observable
|> actorx.filter(is_high_priority)
|> actorx.take_until(shutdown_signal)
|> actorx.subscribe(alert_observer)

Groups elements by key, returning an observable of grouped observables.

Each time a new key is encountered, emits a tuple of (key, Observable). All elements with that key are forwarded to the corresponding group’s observable.

Example

// Group numbers by even/odd
from_list([1, 2, 3, 4, 5, 6])
|> group_by(fn(x) { x % 2 })
|> flat_map(fn(group) {
  let #(key, values) = group
  values |> map(fn(v) { #(key, v) })
})

Creates an observable that emits incrementing integers (0, 1, 2, …) at regular intervals.

Take only the last element. Errors if source is empty.

Create an observer that only handles OnNext events.

Create an observer from three callback functions.

Transform each element using a mapper function.

Transform each element using a mapper function that also receives the index.

Example

from_list(["a", "b", "c"])
|> mapi(fn(x, i) { #(i, x) })
// Emits: #(0, "a"), #(1, "b"), #(2, "c")

Merges multiple observable sequences into a single observable sequence. Values are emitted as they arrive from any source. Completes when all sources complete.

Merge two observables.

Flattens an Observable of Observables by merging inner emissions.

Subscribes to each inner observable as it arrives and forwards all emissions. Completes when outer AND all inners complete.

Create an observable that never emits and never completes.

Forward a notification to an observer.

Send an OnCompleted notification to an observer.

Send an OnError notification to an observer.

Send an OnNext notification to an observer.

Emits consecutive pairs of values.

Example

from_list([1, 2, 3, 4])
|> pairwise()
// Emits: #(1, 2), #(2, 3), #(3, 4)

Converts a cold observable into a connectable hot observable.

Returns a tuple of (Observable, connect_fn) where:

• The Observable can be subscribed to by multiple observers
• The connect function starts the source subscription

Values are only emitted after connect() is called. Multiple subscribers share the same source subscription.

Example

let #(hot, connect) = publish(cold_source)

// Subscribe multiple observers (source not started yet)
let _d1 = hot |> actorx.subscribe(observer1)
let _d2 = hot |> actorx.subscribe(observer2)

// Now connect - source starts, both observers receive values
let connection = connect()

Alias for amb - returns the observable that emits first.

Applies an accumulator function over the source, emitting only the final accumulated value when the source completes.

Example

from_list([1, 2, 3, 4, 5])
|> reduce(0, fn(acc, x) { acc + x })
// Emits: 15, then completes

Resubscribes to the source observable when an error occurs, up to the specified number of retries.

Example

// Retry up to 3 times on error
flaky_observable
|> retry(3)

Sample source when sampler emits.

Applies an accumulator function over the source, emitting each intermediate result.

Example

from_list([1, 2, 3, 4, 5])
|> scan(0, fn(acc, x) { acc + x })
// Emits: 1, 3, 6, 10, 15

Shares a single subscription to the source among multiple subscribers.

Automatically connects when the first subscriber subscribes, and disconnects when the last subscriber unsubscribes.

This is equivalent to publish(source) with automatic reference counting.

Example

let shared =
  interval(100)
  |> share()

// First subscriber - source starts
let d1 = shared |> actorx.subscribe(observer1)

// Second subscriber - shares same source
let d2 = shared |> actorx.subscribe(observer2)

Create an observable that emits a single value then completes.

Creates a single-subscriber subject.

Returns a tuple of (Observer, Observable) where:

• The Observer side is used to push notifications
• The Observable side can be subscribed to (once only!)

Skip the first N elements.

Skip elements while predicate returns True.

Prepends values before the source emissions.

Creates a multicast subject that allows multiple subscribers.

Returns a tuple of (Observer, Observable) where:

• The Observer side is used to push notifications
• The Observable side can be subscribed to by multiple observers

Unlike single_subject, notifications are NOT buffered.

Subscribe an observer to an observable.

Flattens an Observable of Observables by switching to the latest.

When a new inner arrives, the previous inner is cancelled. Useful for search-as-you-type patterns.

Project each element to an observable and switch to the latest.

Composed from map and switch_inner: switch_map(source, f) = source |> map(f) |> switch_inner()

Project each element and its index to an observable and switch to latest.

Take the first N elements.

Take the last N elements (emitted on completion).

Take elements until another observable emits.

Take elements while predicate returns True.

Performs a side effect for each emission without transforming.

Rate limits emissions to at most one per specified period.

Errors if no emission occurs within the specified timeout period. The timeout resets after each emission.

Creates an observable that emits 0 after the specified delay, then completes.

Sends each emitted value to a Subject while passing through to downstream.

Only OnNext values are sent to the Subject. OnError and OnCompleted are forwarded downstream but not sent to the Subject.

Example

actorx.interval(100)
|> actorx.take(10)
|> actorx.map(fn(n) { Increment(n) })
|> actorx.to_subject(counter_inbox)
|> actorx.subscribe(log_observer)

Combines the source observable with the latest value from another observable. Emits only when the source emits, pairing with the latest value from sampler. Completes when source completes.

Combines two observable sequences by pairing their elements by index. Emits when both sources have emitted a value at the same index. Completes when either source completes.