A Go implementation of the ReactiveX spec.

The purpose of Reactive Programming is to simplify the development of event-driven and asynchronous applications by providing a declarative and composable way to handle streams of data or events.

See also:

• samber/lo: A Lodash-style Go library based on Go 1.18+ Generics
• samber/do: A dependency injection toolkit based on Go 1.18+ Generics
• samber/mo: Monads based on Go 1.18+ Generics (Option, Result, Either...)
• samber/cc-skills-golang: AI Agent Skills for Golang

What makes it different from samber/lo?

• lo: synchronous helpers across finite sequences (maps, slices...)
• ro: processing of infinite data streams for event-driven scenarios

Reactive Programming is focused on handling asynchronous data streams where values (like user input, API responses, or sensor data) are emitted over time. Instead of pulling data or waiting for events manually, you react to changes as they occur using Observable, Observer, and Operator. This approach simplifies building systems that are responsive, resilient, and scalable, especially in event-driven or real-time applications.

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)
    }),
)

// Start consuming on subscription
observable.Subscribe(ro.NewObserver(
    func(s string) { fmt.Println(s) },
    func(err error) { fmt.Println(err.Error()) },
    func() { fmt.Println("Completed!") }
))
// Output:
//   "even-0"
//   "even-2"
//   "even-4"
//   "Completed!"

// or:

values, err := ro.Collect(observable)
// []string{"even-0", "even-2", "even-4"}
// <nil>

Now you discovered the paradigm, follow the documentation and turn reactive: 🚀 Getting started

go get github.com/samber/ro

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

This library is v0 and follows SemVer strictly.

Some breaking changes might be made to exported APIs before v1.0.0. Experimental packages under exp/ are considered as unstable.

This library has minimal dependencies outside the Go standard library.

Full documentation here.

The ro library provides all basic operators:

• Creation operators: The data source, usually the first argument of ro.Pipe
• Chainable operators: They filter, validate, transform, enrich... messages
  • Transforming operators: They transform items emitted by an Observable
  • Filtering operators: They selectively emit items from a source Observable
  • Conditional operators: Boolean operators
  • Math and aggregation operators: They perform basic math operations
  • Error handling operators: They help to recover from error notifications from an Observable
  • Combining operators: Combine multiple Observable into one
  • Connectable operators: Convert cold into hot Observable
  • Other: manipulation of context, utility, async scheduling...
• Plugins: External operators (mostly IOs and library wrappers)

The ro library provides a rich ecosystem of plugins for various use cases:

Full documentation here.

• Bytes (plugins/bytes) - String and byte slice manipulation operators
• Strings (plugins/strings) - String manipulation operators
• Sort (plugins/sort) - Sorting operators
• Type Conversion (plugins/strconv) - String conversion operators
• SIMD (plugins/exp/simd) - SIMD-accelerated transformations

• JSON (plugins/encoding/json) - JSON marshaling and unmarshaling
• CSV (plugins/encoding/csv) - CSV reading and writing
• Base64 (plugins/encoding/base64) - Base64 encoding and decoding
• Gob (plugins/encoding/gob) - Go binary serialization

• Cron (plugins/cron) - Schedule jobs using cron expressions or duration intervals
• ICS (plugins/ics) - Read and parse ICS/iCal calendars

• HTTP (plugins/http) - HTTP request operators
• I/O (plugins/io) - File and stream I/O operators
• File System (plugins/fsnotify) - File system monitoring operators

• Log (plugins/observability/log) - Standard logging operators
• Zap (plugins/observability/zap) - Structured logging with zap
• Logrus (plugins/observability/logrus) - Structured logging with logrus
• Slog (plugins/observability/slog) - Structured logging with slog
• Zerolog (plugins/observability/zerolog) - Structured logging with zerolog
• Sentry (plugins/observability/sentry) - Error tracking with Sentry
• Oops (plugins/samber/oops) - Structured error handling

• Native (plugins/ratelimit/native) - Native rate limiting operators
• Ulule (plugins/ratelimit/ulule) - Rate limiting with ulule/limiter

• Regular Expressions (plugins/regexp) - Regular expression operators
• Templates (plugins/template) - Template processing operators

• Process (plugins/proc) - Process execution operators
• Signal (plugins/signal) - Signal handling operators
• Iterators (plugins/iter) - Iterator operators
• PSI (plugins/samber/psi) - Starvation notifier

• Validation (plugins/ozzo/ozzo-validation) - Data validation operators

• Testing (plugins/testify) - Testing utilities

• HyperLogLog (plugins/hyperloglog) - Cardinality estimation operators
• Hot (plugins/samber/hot) - In-memory cache

• Documentation - Official doc
• Godoc - API Reference
• Plugins - Individual plugin documentation
• Examples - Working examples

See the examples directory for complete working examples:

• Stocker price enrichment - Demonstrate a websocket client with data enrichment
• Connectable - Demonstrates connectable observables
• Distributed WebSocket Gateway - Shows how to build a distributed WebSocket gateway
• Parallel API Requests - Demonstrates concurrent HTTP requests
• SQL to CSV - Shows how to process database results to CSV
• ICS to CSV - Shows how to process ICS calendar files to CSV format
• Enterprise Edition Examples - Examples using enterprise features

Check the contribution guide.

• Ping me on Twitter @samuelberthe (DMs, mentions, whatever :))
• Fork the project
• Fix open issues or request new features

Don't hesitate ;)

Give a ⭐️ if this project helped you!

Copyright © 2025 Samuel Berthe.

This project is licensed under the Apache 2.0 License - see the LICENSE file for details.

Note: The ee/ directory contains the Enterprise Edition of the library, which is subject to a custom license. Please refer to the ee/LICENSE.md file for the specific terms and conditions applicable to the Enterprise Edition.