samber/ro

ro - Streams & reactive programming for Go

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

The Reactive Programming paradigm

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

🚀 Install

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.

Core package

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)

Plugins

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

Full documentation here.

Data Manipulation

• 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

Encoding & Serialization

• 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

Scheduling & Timing

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

Network & I/O

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

Observability & Logging

• 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

Rate Limiting

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

Text Processing

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

System Integration

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

Data Validation

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

Testing

• Testing (plugins/testify) - Testing utilities

Utilities

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

📚 Documentation

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

👀 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

🤝 Contributing

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 ;)

👤 Contributors

💫 Show your support

Give a ⭐️ if this project helped you!

License

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.