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. ---- 💖 Sponsored by: DBOS - Durable workflow orchestration library for Go ---- **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 , , and . This approach simplifies building systems that are responsive, resilient, and scalable, especially in event-driven or real-time applications. Now you discovered the paradigm, follow the documentation and turn reactive: 🚀 Getting started 🚀 Install This library is v0 and follows SemVer strictly. Some breaking changes might be made to exported APIs before v1.0.0. Experimental packages under are considered as unstable. This library has minimal dependencies outside the Go standard library. Core package Full documentation here. The library provides all basic operators: • **Creation operators**: The data source, usually the first argument of • **Chainable operators**: They filter, validate, transform, enrich... messages • **Transforming operators**: They transform items emitted by an • **Filtering operators**: They selectively emit items from a source • **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 • **Combining operators**: Combine multiple into one • **Connectable operators**: Convert cold into hot • **Other**: manipulation of context, utility, async scheduling... • **Plugins**: External operators (mostly IOs and library wrappers) Plugins The library provides a rich ecosystem of plugins for various use cases: Full documentation here. Data Manipulation • **Bytes** ( ) - String and byte slice manipulation operators • **Strings** ( ) - String manipulation operators • **Sort** ( ) - Sorting operators • **Type Conversion** ( ) - String conversion operators • **SIMD** ( ) - SIMD-accelerated transformations Encoding & Serialization • **JSON** ( ) - JSON marshaling and unmarshaling • **CSV** ( ) - CSV reading and writing • **Base64** ( ) - Base64 encoding and decoding • **Gob** ( ) - Go binary serialization Scheduling & Timing • **Cron** ( ) - Schedule jobs using cron expressions or duration intervals • **ICS** ( ) - Read and parse ICS/iCal calendars Network & I/O • **HTTP** ( ) - HTTP request operators • **I/O** ( ) - File and stream I/O operators • **File System** ( ) - File system monitoring operators Observability & Logging • **Log** ( ) - Standard logging operators • **Zap** ( ) - Structured logging with zap • **Logrus** ( ) - Structured logging with logrus • **Slog** ( ) - Structured logging with slog • **Zerolog** ( ) - Structured logging with zerolog • **Sentry** ( ) - Error tracking with Sentry • **Oops** ( ) - Structured error handling Rate Limiting • **Native** ( ) - Native rate limiting operators • **Ulule** ( ) - Rate limiting with ulule/limiter Text Processing • **Regular Expressions** ( ) - Regular expression operators • **Templates** ( ) - Template processing operators System Integration • **Process** ( ) - Process execution operators • **Signal** ( ) - Signal handling operators • **Iterators** ( ) - Iterator operators • **PSI** ( ) - Starvation notifier Data Validation • **Validation** ( ) - Data validation operators Testing • **Testing** ( ) - Testing utilities Utilities • **HyperLogLog** ( ) - Cardinality estimation operators • **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 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.