RxV is a ReactiveX implementation for the V programming language — fully generic, built on channels, zero dependencies.

Table of Contents

• What is ReactiveX?
• Install
• Quick Start
• Supported Operators
• V Compiler Notes
• Examples
• Tutorials
• API Reference
• Testing
• Contributing

What is ReactiveX?

ReactiveX is an API for programming with Observable streams. It provides a unified model for handling asynchronous data — whether from user events, HTTP responses, database results, or any other source.

RxV brings this model to V using:

• Generic Observables — ObservableImpl[T] works with any type
• Channel-based pipelines — each operator spawns a thread and connects via chan Item[T]
• Composable operators — filter, map, merge, reduce, and more

Install

v install ulises-jeremias.rxv

Quick Start

import ulises_jeremias.rxv

fn main() {
    // 1. Create a stream of integers
    mut obs := rxv.range(1, 6) // emits 1, 2, 3, 4, 5

    // 2. Keep only even numbers
    mut evens := obs.filter(fn (v int) bool {
        return v % 2 == 0
    })

    // 3. Double each value using map_ (free function — see V Compiler Notes)
    mut doubled := rxv.map_[int, int](mut evens, fn (v int) ?int {
        return v * 2
    })

    // 4. Subscribe and collect results
    done := doubled.for_each(fn (v int) {
        println(v)
    }, fn (e IError) {
        eprintln('error: ${e}')
    }, fn () {
        println('done')
    })
    _ = <-done
    // Output:
    // 4
    // 8
    // done
}

A More Complete Example

import ulises_jeremias.rxv

fn main() {
    // Aggregate: sum all integers from 1 to 10
    mut obs := rxv.range(1, 10)
    mut total := rxv.reduce_[int, int](mut obs, 0, fn (acc int, val int) int {
        return acc + val
    })

    done := total.for_each(fn (v int) {
        println('Sum 1..10 = ${v}')
    }, fn (e IError) {
        eprintln('error: ${e}')
    }, fn () {})
    _ = <-done
    // Output: Sum 1..10 = 55
}

Supported Operators

Creating

Filtering

Timing (free functions — see V Compiler Notes)

Transforming (free functions — see V Compiler Notes)

Aggregating (free functions)

Combining

Mathematical (f64 only)

Subscribing

V Compiler Notes

RxV targets V 0.5.x. The current compiler has limitations with certain generic patterns:

Methods cannot have additional type parameters.
(mut o ObservableImpl[T]) map[U](fn(T) U) is not supported.

Workaround: Operators that transform to a different type are exposed as free functions named with a _ suffix:

// Instead of obs.map[string](fn(v int) string { ... })
mut labels := rxv.map_[int, string](mut obs, fn (v int) ?string { return v.str() })

// Instead of obs.scan[int](0, accumulator)
mut running := rxv.scan_[int, int](mut obs, 0, fn (acc int, v int) int { return acc + v })

See the full list of compiler workarounds in docs/API.md.

Examples

Run any example:

v run examples/05-aggregation/main.v

Tutorials

Step-by-step guides in docs/tutorials/:

1. Hello World
2. Creating Observables
3. Filtering
4. Transforming
5. Error Handling

API Reference

→ docs/README.md — documentation home → docs/API.md — full type and function reference → docs/OPERATORS.md — operator reference with examples

Testing

./bin/test

Or run a single test file directly:

cd ~/.vmodules/ulises_jeremias && v run rxv/filter_observe_test.v

Note: Avoid v test rxv — it runs all tests in parallel which can
exhaust system resources. Use ./bin/test which runs them sequentially.

Contributing

See CONTRIBUTING.md for the contribution workflow.

[workflowbadge]: https://github.com/ulises-jeremias/rxv/actions/workflows/ci.yml/badge.svg [deploydocsbadge]: https://github.com/ulises-jeremias/rxv/actions/workflows/deploy-docs.yml/badge.svg [licensebadge]: https://img.shields.io/badge/License-MIT-blue.svg [workflowurl]: https://github.com/ulises-jeremias/rxv/actions/workflows/ci.yml [deploydocsurl]: https://github.com/ulises-jeremias/rxv/actions/workflows/deploy-docs.yml [licenseurl]: https://github.com/ulises-jeremias/rxv/blob/main/LICENSE

fn concat[T](observables []&ObservableImpl[T], opts ...RxOption) &ObservableImpl[T]

concat concatenates multiple observables sequentially.

fn concat_map_[T, U](mut o ObservableImpl[T], mapper fn (t T) &ObservableImpl[U], opts ...RxOption) &ObservableImpl[U]

concat_map_ transforms each item by applying a function that returns an observable, and flattens them sequentially (one inner completes before the next starts).

fn count_[T](mut o ObservableImpl[T], opts ...RxOption) &ObservableImpl[int]

count_ returns an Observable emitting the number of items emitted by the source.

fn create[T](producer ProducerFn[T]) &ObservableImpl[T]

create builds an ObservableImpl from a producer function.

fn debounce_[T](mut o ObservableImpl[T], delay_ms int, opts ...RxOption) &ObservableImpl[T]

debounce emits an item only after the specified delay has passed without any other item.

fn defer_[T](factory fn () &ObservableImpl[T]) &ObservableImpl[T]

fn empty[T]() &ObservableImpl[T]

empty completes immediately without emitting any item.

fn flat_map_[T, U](mut o ObservableImpl[T], mapper fn (t T) &ObservableImpl[U], opts ...RxOption) &ObservableImpl[U]

flat_map_ transforms each item by applying a function that returns an observable, then merges all inner observables into a single stream.

fn from_channel[T](ch chan Item[T]) &ObservableImpl[T]

from_channel wraps an existing channel as an ObservableImpl.

fn from_error[T](err IError) Item[T]

fn from_slice[T](items []T) &ObservableImpl[T]

fn interval(period_ms int) &ObservableImpl[int]

interval emits sequential integers starting at 0 at each period_ms milliseconds. The returned observable never completes — use take() to limit emissions.

fn just[T](items ...T) &ObservableImpl[T]

just emits each provided value then completes.

fn map_[T, U](mut o ObservableImpl[T], apply MapFn[T, U], opts ...RxOption) &ObservableImpl[U]

fn merge[T](mut o1 ObservableImpl[T], mut o2 ObservableImpl[T], opts ...RxOption) &ObservableImpl[T]

merge combines two observables by interleaving their emissions.

fn new_channel_iterable[T](ch chan Item[T]) &ChannelIterable[T]

fn new_create_iterable[T](producer ProducerFn[T]) &CreateIterable[T]

fn new_slice_iterable[T](items []T) &SliceIterable[T]

fn of[T](value T) Item[T]

fn parse_options(opts ...RxOption) Options

parse_options merges a list of RxOption functions into an Options value.

fn range(start int, count int) &ObservableImpl[int]

range emits integers in [start, start+count).

fn reduce_[T, U](mut o ObservableImpl[T], seed U, accumulator fn (acc U, val T) U, opts ...RxOption) &ObservableImpl[U]

reduce_ applies an accumulator over the entire stream, emitting only the final value.

fn repeat[T](value T, count int) &ObservableImpl[T]

repeat emits the same value count times.

fn sample[T](mut o ObservableImpl[T], period_ms int, opts ...RxOption) &ObservableImpl[T]

sample emits the most recent item at the specified periodic interval.

fn scan_[T, U](mut o ObservableImpl[T], seed U, accumulator fn (acc U, val T) U, opts ...RxOption) &ObservableImpl[U]

scan_ applies an accumulator over the stream, emitting each intermediate result.

fn throttle_first_[T](mut o ObservableImpl[T], delay_ms int, opts ...RxOption) &ObservableImpl[T]

throttle_first emits the first item, then ignores subsequent items until delay expires.

fn throw[T](err IError) &ObservableImpl[T]

throw emits a single error then completes.

fn timer(delay_ms int) &ObservableImpl[int]

timer emits a single 0 after delay_ms milliseconds, then completes.

fn with_buffer_size(n int) RxOption

--- built-in option constructors ---

fn with_context(ctx context.Context) RxOption

fn with_eager_observation() RxOption

fn with_error_strategy(strategy OnErrorStrategy) RxOption

fn with_pool(n int) RxOption

fn BackpressureStrategy.from[W](input W) !BackpressureStrategy

fn CloseChannelStrategy.from[W](input W) !CloseChannelStrategy

fn ObservationStrategy.from[W](input W) !ObservationStrategy

fn OnErrorStrategy.from[W](input W) !OnErrorStrategy

interface Iterable[T] {
mut:
	observe(opts ...RxOption) chan Item[T]
}

Iterable[T] is the base interface for anything that can be observed.

interface Observable[T] {
	Iterable
}

Note: Observable[T] as a generic interface with generic method params (likefilter(PredicateFn[T])) is not yet fully supported by the V compiler — the interface check fails to substitute T in method signatures. We therefore expose ObservableImpl[T] directly as the public API. The Observable[T] interface is kept minimal (only observe) for use where a common supertype is needed without operators.

fn (mut i ChannelIterable[T]) observe(opts ...RxOption) chan Item[T]

type CompletedFn = fn ()

CompletedFn is called when the stream completes.

fn (mut i CreateIterable[T]) observe(opts ...RxOption) chan Item[T]

type ErrFn = fn (err IError)

ErrFn is called when an error item is received.

type MapFn[T, U] = fn (value T) ?U

MapFn[T, U] transforms a T value into a U, or returns an error.

type NextFn[T] = fn (value T)

NextFn[T] is called for each next value in for_each.

fn (o ObservableImpl[T]) str() string

fn (mut o ObservableImpl[f64]) average_f64(opts ...RxOption) &ObservableImpl[f64]

average_f64 computes the average of f64 values emitted by the source.

fn (mut o ObservableImpl[f64]) sum_f64(opts ...RxOption) &ObservableImpl[f64]

sum_f64 computes the sum of f64 values emitted by the source.

type PredicateFn[T] = fn (value T) bool

PredicateFn[T] tests a value; return true to keep it.

type ProducerFn[T] = fn (mut ctx context.Context, ch chan Item[T])

ProducerFn[T] sends items onto a channel given a context.

type RxOption = fn (mut options Options)

RxOption is a functional option applied to an Options struct.

fn (mut i SliceIterable[T]) observe(opts ...RxOption) chan Item[T]

fn (i Item[T]) is_error() bool

fn (i Item[T]) get_value() T

fn (i Item[T]) send_context(done chan int, ch chan Item[T]) bool

fn (mut o ObservableImpl[T]) observe(opts ...RxOption) chan Item[T]

fn (mut o ObservableImpl[T]) filter(predicate PredicateFn[T], opts ...RxOption) &ObservableImpl[T]

fn (mut o ObservableImpl[T]) take(n u32, opts ...RxOption) &ObservableImpl[T]

fn (mut o ObservableImpl[T]) for_each(next_fn NextFn[T], err_fn ErrFn, completed_fn CompletedFn, opts ...RxOption) chan int

fn (mut o ObservableImpl[T]) distinct(opts ...RxOption) &ObservableImpl[T]

distinct suppresses duplicate items, emitting only items not previously seen.

fn (mut o ObservableImpl[T]) distinct_until_changed(opts ...RxOption) &ObservableImpl[T]

distinct_until_changed suppresses consecutive duplicate items.

fn (mut o ObservableImpl[T]) first(opts ...RxOption) &ObservableImpl[T]

first emits only the first item from the source.

fn (mut o ObservableImpl[T]) last(opts ...RxOption) &ObservableImpl[T]

last emits only the last item from the source.

fn (mut o ObservableImpl[T]) skip(n u32, opts ...RxOption) &ObservableImpl[T]

skip suppresses the first n items emitted by the source.

fn (mut o ObservableImpl[T]) take_last(n u32, opts ...RxOption) &ObservableImpl[T]

take_last emits only the last n items from the source.

fn (mut o ObservableImpl[T]) contains(pred PredicateFn[T], opts ...RxOption) &ObservableImpl[bool]

contains returns true if the source emits an item that satisfies the predicate.

fn (mut o ObservableImpl[T]) is_empty(opts ...RxOption) &ObservableImpl[bool]

is_empty returns true if the source Observable completes without emitting any item.

fn (mut o ObservableImpl[T]) element_at(index u32, opts ...RxOption) &ObservableImpl[T]

element_at returns the item at the specified index or errors if out of bounds.

fn (mut o ObservableImpl[T]) all(pred PredicateFn[T], opts ...RxOption) &ObservableImpl[bool]

all returns true if all items satisfy the predicate.

fn (mut o ObservableImpl[T]) any(pred PredicateFn[T], opts ...RxOption) &ObservableImpl[bool]

any returns true if at least one item satisfies the predicate.

fn (mut o ObservableImpl[T]) find(pred PredicateFn[T], opts ...RxOption) &ObservableImpl[T]

find returns the first item that satisfies the predicate.

fn (mut o ObservableImpl[T]) timeout(timeout_ms int, opts ...RxOption) &ObservableImpl[T]

timeout emits an error if no item is received within timeout_ms milliseconds.

fn (o Options) build_channel_t[T]() chan Item[T]

build_channel_t creates a typed buffered channel.

fn (o Options) get_pool() ?int

get_pool returns the pool size if set.

fn (o Options) is_eager_observation() bool

enum BackpressureStrategy {
	block
	drop
}

BackpressureStrategy controls how full channels are handled.

enum CloseChannelStrategy {
	leave_channel_open
	close_channel
}

enum ObservationStrategy {
	lazy
	eager
}

ObservationStrategy controls eager vs lazy subscription.

enum OnErrorStrategy {
	stop_on_error
	continue_on_error
}

OnErrorStrategy controls operator behaviour on errors.

struct ChannelIterable[T] {
mut:
	next chan Item[T]
}

---- ChannelIterable -------------------------------------------------------- Used internally to get a channel from a pre-existing channel.

struct CreateIterable[T] {
mut:
	producer ProducerFn[T] = unsafe { nil }
}

---- CreateIterable ---------------------------------------------------------

struct Item[T] {
pub:
	value     T
	has_value bool
	err       IError
}

struct ObservableImpl[T] {
mut:
	ch     chan Item[T]
	parent context.Context
}

struct Options {
mut:
	buffer_size    int
	pool           int
	ctx            context.Context
	eager          bool
	error_strategy OnErrorStrategy
}

Options holds resolved operator configuration.

fn (mut o Options) build_context(parent context.Context) context.Context

build_context returns the context for an operator.

struct SliceIterable[T] {
mut:
	items []T
}

---- SliceIterable ----------------------------------------------------------