ReactiveX (RxPY) Quick Reference

RxPY provides composable asynchronous data streams. This is a practical guide focused on common patterns in this codebase.

Quick Start: Using an Observable

Given a function that returns an Observable, here’s how to use it:

session=rx

import reactivex as rx
from reactivex import operators as ops

# Create an observable that emits 0,1,2,3,4
source = rx.of(0, 1, 2, 3, 4)

# Subscribe and print each value
received = []
source.subscribe(lambda x: received.append(x))
print("received:", received)

received: [0, 1, 2, 3, 4]

The `.pipe()` Pattern

Chain operators using .pipe():

session=rx

# Transform values: multiply by 2, then filter > 4
result = []

# We build another observable. It's passive until `subscribe` is called.
observable = source.pipe(
    ops.map(lambda x: x * 2),
    ops.filter(lambda x: x > 4),
)

observable.subscribe(lambda x: result.append(x))

print("transformed:", result)

transformed: [6, 8]

Common Operators

Transform: `map`

session=rx

rx.of(1, 2, 3).pipe(
    ops.map(lambda x: f"item_{x}")
).subscribe(print)

item_1
item_2
item_3
<reactivex.disposable.disposable.Disposable object at 0x7f5fa44d0d10>

Filter: `filter`

session=rx

rx.of(1, 2, 3, 4, 5).pipe(
    ops.filter(lambda x: x % 2 == 0)
).subscribe(print)

2
4
<reactivex.disposable.disposable.Disposable object at 0x7f5fa44d0c20>

Limit emissions: `take`

session=rx

rx.of(1, 2, 3, 4, 5).pipe(
    ops.take(3)
).subscribe(print)

1
2
3
<reactivex.disposable.disposable.Disposable object at 0x7f5fa44d0e60>

Flatten nested observables: `flat_map`

session=rx

# For each input, emit multiple values
rx.of(1, 2).pipe(
    ops.flat_map(lambda x: rx.of(x, x * 10, x * 100))
).subscribe(print)

1
10
100
2
20
200
<reactivex.disposable.disposable.Disposable object at 0x7f5fa44d1b80>

Rate Limiting

`sample(interval)` - Emit latest value every N seconds

Takes the most recent value at each interval. Good for continuous streams where you want the freshest data.

session=rx

# Use blocking .run() to collect results properly
results = rx.interval(0.05).pipe(
    ops.take(10),
    ops.sample(0.2),
    ops.to_list(),
).run()
print("sample() got:", results)

sample() got: [2, 6, 9]

`throttle_first(interval)` - Emit first, then block for N seconds

Takes the first value then ignores subsequent values for the interval. Good for user input debouncing.

session=rx

results = rx.interval(0.05).pipe(
    ops.take(10),
    ops.throttle_first(0.15),
    ops.to_list(),
).run()
print("throttle_first() got:", results)

throttle_first() got: [0, 3, 6, 9]

Difference Between `sample` and `throttle_first`

session=rx

# sample: takes LATEST value at each interval tick
# throttle_first: takes FIRST value then blocks

# With fast emissions (0,1,2,3,4,5,6,7,8,9) every 50ms:
# sample(0.2s)        -> gets value at 200ms, 400ms marks -> [2, 6, 9]
# throttle_first(0.15s) -> gets 0, blocks, then 3, blocks, then 6... -> [0,3,6,9]
print("sample: latest value at each tick")
print("throttle_first: first value, then block")

sample: latest value at each tick
throttle_first: first value, then block

What is an Observable?

An Observable is like a list, but instead of holding all values at once, it produces values over time.

	List	Iterator	Observable
Values	All exist now	Generated on demand	Arrive over time
Control	You pull (`for x in`)	You pull (`next()`)	Pushed to you
Size	Finite	Can be infinite	Can be infinite
Async	No	Yes (with asyncio)	Yes
Cancel	N/A	Stop calling `next()`	`.dispose()`

The key difference from iterators: with an Observable, you don’t control when values arrive. A camera produces frames at 30fps whether you’re ready or not. An iterator waits for you to call next(). Observables are lazy. An Observable is just a description of work to be done - it sits there doing nothing until you call .subscribe(). That’s when it “wakes up” and starts producing values. This means you can build complex pipelines, pass them around, and nothing happens until someone subscribes. The three things an Observable can tell you:

“Here’s a value” (on_next) - A new value arrived
“Something went wrong” (on_error) - An error occurred, stream stops
“I’m done” (on_completed) - No more values coming

The basic pattern:

observable.subscribe(what_to_do_with_each_value)

That’s it. You create or receive an Observable, then subscribe to start receiving values. When you subscribe, data flows through a pipeline:

Key property: Observables are lazy. Nothing happens until you call .subscribe(). This means you can build up complex pipelines without any work being done, then start the flow when ready. Here’s the full subscribe signature with all three callbacks:

session=rx

rx.of(1, 2, 3).subscribe(
    on_next=lambda x: print(f"value: {x}"),
    on_error=lambda e: print(f"error: {e}"),
    on_completed=lambda: print("done")
)

value: 1
value: 2
value: 3
done
<reactivex.disposable.disposable.Disposable object at 0x7f5fa44d33b0>

Disposables: Cancelling Subscriptions

When you subscribe, you get back a Disposable. This is your “cancel button”:

session=rx

import reactivex as rx

source = rx.interval(0.1)  # emits 0, 1, 2, ... every 100ms forever
subscription = source.subscribe(lambda x: print(x))

# Later, when you're done:
subscription.dispose()  # Stop receiving values, clean up resources
print("disposed")

disposed

Why does this matter?

Observables can be infinite (sensor feeds, websockets, timers)
Without disposing, you leak memory and keep processing values forever
Disposing also cleans up any resources the Observable opened (connections, file handles, etc.)

Rule of thumb: Whenever you subscribe, save the disposable because you have to unsubscribe at some point by calling disposable.dispose(). In dimos modules: Every Module has a self._disposables (a CompositeDisposable) that automatically disposes everything when the module closes:

session=rx

import time

import reactivex as rx
from dimos.core.module import Module

class MyModule(Module):
    def start(self) -> None:
        super().start()
        source = rx.interval(0.05)
        self.register_disposable(source.subscribe(lambda x: print(f"got {x}")))

module = MyModule()
module.start()
time.sleep(0.25)

# unsubscribes disposables
module.stop()

got 0
got 1
got 2
got 3

Creating Observables

There are two common callback patterns in APIs. Use the appropriate helper:

Pattern	Example	Helper
Register/unregister with same callback	`sensor.register(cb)` / `sensor.unregister(cb)`	`callback_to_observable`
Subscribe returns unsub function	`unsub = pubsub.subscribe(cb)`	`to_observable`

From register/unregister APIs

Use callback_to_observable when the API has separate register and unregister functions that take the same callback reference:

session=create

import reactivex as rx
from reactivex import operators as ops
from dimos.utils.reactive import callback_to_observable

class MockSensor:
    def __init__(self):
        self._callbacks = []
    def register(self, cb):
        self._callbacks.append(cb)
    def unregister(self, cb):
        self._callbacks.remove(cb)
    def emit(self, value):
        for cb in self._callbacks:
            cb(value)

sensor = MockSensor()

obs = callback_to_observable(
    start=sensor.register,
    stop=sensor.unregister
)

received = []
sub = obs.subscribe(lambda x: received.append(x))

sensor.emit("reading_1")
sensor.emit("reading_2")
print("received:", received)

sub.dispose()
print("callbacks after dispose:", len(sensor._callbacks))

received: ['reading_1', 'reading_2']
callbacks after dispose: 0

Use to_observable when the subscribe function returns an unsubscribe callable:

session=create

from dimos.utils.reactive import to_observable

class MockPubSub:
    def __init__(self):
        self._callbacks = []
    def subscribe(self, cb):
        self._callbacks.append(cb)
        return lambda: self._callbacks.remove(cb)  # returns unsub function
    def publish(self, value):
        for cb in self._callbacks:
            cb(value)

pubsub = MockPubSub()

obs = to_observable(pubsub.subscribe)

received = []
sub = obs.subscribe(lambda x: received.append(x))

pubsub.publish("msg_1")
pubsub.publish("msg_2")
print("received:", received)

sub.dispose()
print("callbacks after dispose:", len(pubsub._callbacks))

received: ['msg_1', 'msg_2']
callbacks after dispose: 0

From scratch with `rx.create`

session=create

from reactivex.disposable import Disposable

def custom_subscribe(observer, scheduler=None):
    observer.on_next("first")
    observer.on_next("second")
    observer.on_completed()
    return Disposable(lambda: print("cleaned up"))

obs = rx.create(custom_subscribe)

results = []
obs.subscribe(
    on_next=lambda x: results.append(x),
    on_completed=lambda: results.append("DONE")
)
print("results:", results)

cleaned up
results: ['first', 'second', 'DONE']

CompositeDisposable

As we know we can always dispose subscriptions when done to prevent leaks:

session=dispose

import time
import reactivex as rx
from reactivex import operators as ops

source = rx.interval(0.1).pipe(ops.take(100))
received = []

subscription = source.subscribe(lambda x: received.append(x))
time.sleep(0.25)
subscription.dispose()
time.sleep(0.2)

print(f"received {len(received)} items before dispose")

received 2 items before dispose

For multiple subscriptions, use CompositeDisposable:

session=dispose

from reactivex.disposable import CompositeDisposable

disposables = CompositeDisposable()

s1 = rx.of(1,2,3).subscribe(lambda x: None)
s2 = rx.of(4,5,6).subscribe(lambda x: None)

disposables.add(s1)
disposables.add(s2)

print("subscriptions:", len(disposables))
disposables.dispose()
print("after dispose:", disposables.is_disposed)

subscriptions: 2
after dispose: True

Reference

Operator	Purpose	Example
`map(fn)`	Transform each value	`ops.map(lambda x: x * 2)`
`filter(pred)`	Keep values matching predicate	`ops.filter(lambda x: x > 0)`
`take(n)`	Take first n values	`ops.take(10)`
`first()`	Take first value only	`ops.first()`
`sample(sec)`	Emit latest every interval	`ops.sample(0.5)`
`throttle_first(sec)`	Emit first, block for interval	`ops.throttle_first(0.5)`
`flat_map(fn)`	Map + flatten nested observables	`ops.flat_map(lambda x: rx.of(x, x))`
`observe_on(sched)`	Switch scheduler	`ops.observe_on(pool_scheduler)`
`replay(n)`	Cache last n values for late subscribers	`ops.replay(buffer_size=1)`
`timeout(sec)`	Error if no value within timeout	`ops.timeout(5.0)`

See RxPY documentation for complete operator reference.

Getting Started

Installation

Usage

Capabilities

Platforms

Development

ReactiveX (RxPY) Quick Reference

Quick Start: Using an Observable

The `.pipe()` Pattern

Common Operators

Transform: `map`

Filter: `filter`

Limit emissions: `take`

Flatten nested observables: `flat_map`

Rate Limiting

`sample(interval)` - Emit latest value every N seconds

`throttle_first(interval)` - Emit first, then block for N seconds

Difference Between `sample` and `throttle_first`

What is an Observable?

Disposables: Cancelling Subscriptions

Creating Observables

From register/unregister APIs

From scratch with `rx.create`

CompositeDisposable

Reference

​Quick Start: Using an Observable

​The .pipe() Pattern

​Common Operators

​Transform: map

​Filter: filter

​Limit emissions: take

​Flatten nested observables: flat_map

​Rate Limiting

​sample(interval) - Emit latest value every N seconds

​throttle_first(interval) - Emit first, then block for N seconds

​Difference Between sample and throttle_first

​What is an Observable?

​Disposables: Cancelling Subscriptions

​Creating Observables

​From register/unregister APIs

​From subscribe-returns-unsub APIs

​From scratch with rx.create

​CompositeDisposable

​Reference

Quick Start: Using an Observable

The `.pipe()` Pattern

Common Operators

Transform: `map`

Filter: `filter`

Limit emissions: `take`

Flatten nested observables: `flat_map`

Rate Limiting

`sample(interval)` - Emit latest value every N seconds

`throttle_first(interval)` - Emit first, then block for N seconds

Difference Between `sample` and `throttle_first`

What is an Observable?

Disposables: Cancelling Subscriptions

Creating Observables

From register/unregister APIs

From subscribe-returns-unsub APIs

From scratch with `rx.create`

CompositeDisposable

Reference