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