ro: reactive streams and a ReactiveX implementation for Go

ReactiveX, in Go, at version zero

ro is a Go implementation of the ReactiveX specification, and the readme puts the purpose plainly: reactive programming exists to simplify the development of event-driven and asynchronous applications by providing a declarative and composable way to handle streams of data or events.

The paradigm it implements is about values that arrive over time, whether that is user input, an API response, or sensor data, and the point of the library is that you react to them as they occur rather than pulling the data yourself or waiting on events by hand. The three nouns in the model are the observable, the observer, and the operator, and the composition happens in a pipe.

The first example in the readme shows all three. A source emits a range of integers at one second intervals, a filter keeps the even ones, and a map turns them into strings.

observable := ro.Pipe(
    ro.RangeWithInterval(0, 5, 1*time.Second),
    ro.Filter(func(x int) bool {
        return x%2 == 0
    }),
    ro.Map(func(x int) string {
        return fmt.Sprintf("even-%d", x)
    }),
)

Then a subscription starts the flow, with three callbacks: one for the values, one for the error, and one for completion.

observable.Subscribe(ro.NewObserver(
    func(s string) { fmt.Println(s) },
    func(err error) { fmt.Println(err.Error()) },
    func() { fmt.Println("Completed!") }
))

The output is the even values followed by the completion message, and the alternative is to collect the whole stream into a slice, which returns the values and a nil error.

There is a second family of projects by the same author that clarifies the scope. The Lodash style library covers synchronous helpers across finite sequences such as maps and slices, while ro covers the processing of infinite data streams for event-driven scenarios. The split is the difference between working on a collection you have and working on a sequence that has not ended.

The project is Apache 2.0 licensed, requires Go 1.18 or newer, has minimal dependencies outside the standard library, and the last push was on 2026-10-19 with releases at v0.4.1, v0.4.0, and v0.3.0 through 2026.

What version zero means for an API

The readme states the compatibility position in two sentences, and they are the first thing to read before you adopt the library.

The library is at version zero and follows semantic versioning strictly. Some breaking changes might still be made to the exported APIs before version 1.0.0. And the packages under the experimental directory are considered unstable.

There is a tension in that, and it is worth naming. Strict semantic versioning is a promise about how versions relate to each other, and honouring it strictly while the major version is zero is a way of saying the versioning discipline is real but the shape of the API is not settled. So a minor release can still move things, and the packages that are explicitly experimental can move at any time.

For a team that means two decisions. Pin the dependency the way you would pin a pre-1.0 library, by minor version, and read the changelog on each upgrade. And keep experimental packages out of the code that other people depend on.

The dependency situation is otherwise good. The module requires the author's own Lodash style library, the test assertion library, a goroutine leak detector, and the experimental package of the Go project, and the readme says the library has minimal dependencies outside the standard library. The leak detector is the one to notice, because it is the tool that catches the failure mode reactive code is most prone to.

The installation is one command, and there is a second command that installs the same library as a skill for coding agents.

go get github.com/samber/ro

# AI Agent Skill
npx skills add https://github.com/samber/cc-skills-golang --skill golang-samber-ro

That second line is part of a set of agent skills the author publishes for the Golang ecosystem, alongside the other libraries in the family.

The operator families

The core package is organised by what an operator does to a stream, and the readme lists the groups, which is the fastest way to understand the surface.

The creation operators are the data source, and they are usually the first argument of the pipe. Everything else is chainable, and the chainable group divides into the transforming operators, which transform items emitted by an observable, and the filtering operators, which selectively emit items from a source. There are conditional operators for the boolean cases, and math and aggregation operators for the arithmetic.

Error handling is its own group, and it exists because an error notification from an observable is a different event from a value, and a pipeline that has to recover from one needs a place to put the recovery. The combining operators merge several observables into one, which is how you join two event sources, and the connectable operators convert a cold observable into a hot one, which is how you share one subscription between several consumers.

The last core group is everything else: manipulation of the context, utility operators, and asynchronous scheduling.

On top of that sits the plugin group, described as external operators, mostly input and output wrappers and library wrappers, and that phrase is the most honest description of what most of the ecosystem is.

The file layout in the repository mirrors the grouping exactly, with one file per operator family at the root of the package: a file for creation, one for transformations, one for filtering, one for conditional, one for combining, one for connectable, one for error handling, one for math, one for sink, one for context, and one for utility. Each has its test file next to it, which is the layout a project with this much surface uses so that a reviewer can find the tests for a group without hunting.

Alongside those are the core types: the observable, the observer, the pipe, the scheduler, and several subject implementations, one of them asynchronous and one behaving like a stateful subject.

The plugin set, which is most of the library

The plugins are grouped by domain, and the grouping tells you what a reactive library turns into when it is applied to real systems rather than to numbers.

For data manipulation there are operators for byte slices and strings, for sorting, and for string to number conversion, plus an experimental package of SIMD accelerated transformations. Encoding covers JSON, CSV, Base64, and Go's own binary format. Scheduling covers cron expressions and duration intervals, and calendar files, so an observable can be driven by a schedule or by an incoming calendar event.

The network and input group is the one that turns the library into an event-driven application rather than a maths exercise. There are HTTP request operators, file and stream input and output operators, and file system monitoring built on the notification library, which means a change on disk can be an observable that other operators transform.

Observability has the widest coverage, with adapters for the standard library logger and for the four common structured logging libraries, plus error tracking through Sentry and a structured error handling plugin of the author's own. Rate limiting exists twice, once with a native implementation and once built on the ulule limiter library. Text processing covers regular expressions and templates. System integration covers process execution, signal handling, iterators, and a starvation notifier.

Then there is the smaller set: data validation through the ozzo library, testing utilities for the testify library, cardinality estimation with HyperLogLog, and an in-memory cache.

The examples directory is the best guide to what these are for, because the directory names are use cases rather than API names. There is a connectable example, a distributed websocket gateway, three examples for the enterprise telemetry packages covering logs, metrics, and traces, a Prometheus example, a converter from calendar files to CSV, a parallel API requests example, a SQL to CSV example, and a stock price enrichment example.

That last category is the point. A pipeline that takes a row from a database, enriches it with a price, and writes the result out is exactly the shape of application code that channels and goroutines make awkward, and it is why a reactive library with a plugin per data source earns its dependency.

Testing reactive code, and the fuzz targets

The build file is where the project explains how it tests itself, and the explanation is unusually specific about the costs.

The ordinary test target runs the whole module with the race detector, and there is a separate command that runs the SIMD plugin when the SIMD experiment is enabled in the Go toolchain.

The fuzz target is the interesting one. Every fuzz target is run on its seeds as part of the normal test run, and the number of seeds is configurable through an environment variable, with a default of a thousand. The comment explains the timing: the fuzz package needs about ninety seconds per hundred seeds under the race detector, so a thousand seeds take about fifteen minutes, which is why the per package timeout for that target is set to thirty minutes rather than the default ten. The comment also notes that the skip flag for excluding fuzz targets from the ordinary run does not exist before Go 1.20, while the CI matrix starts at 1.18, so the seeds run in both.

There is a watch mode that polls every fifty milliseconds and runs the tests through a helper, and a benchmark target that runs the benchmarks verbosely with memory allocation reporting and three repetitions, so a change in allocation is visible rather than averaged away.

The presence of a goroutine leak detector in the dependencies is the piece that matters most for a reactive library, because the classic defect in this style of code is a subscription that is never cancelled. A pipeline that holds a reference to a source forever will not fail a functional test, it will simply keep a goroutine alive, and a leak detector turns that into a test failure.

The repository also carries a workspace file, a citation file, a linter configuration, and separate instruction files for coding agents, and a directory of licence texts for the code that has been vendored or derived.

Editorial conclusion

ro fits a Go developer building an event-driven service, where several goroutines consume the same events, back pressure matters, and the alternative is a tangle of channels and context cancellations. It does not fit a team that wants a stable library, because the project is at version zero and says so, and it does not fit a project that has standardised on channels, since that is the idiom Go gives you for free. Verify first that the operators you need exist in the core package or in a plugin rather than in the experimental directory, since the packages under exp are declared unstable, that your Go version is one the library supports, and that the goroutine leak detector is enabled in your own tests, because a reactive pipeline that leaves a subscription running is the failure mode this library's test tooling is built to catch.

Frequently asked questions

Official sources

1. License: Apache-2.0
2. Project website
3. README
4. Releases
5. samber/ro on GitHub