ro - Reactive programming for Go

website·Indexed 33 days ago

https://ro.samber.dev

A Go implementation of the ReactiveX specification providing a declarative and composable way to handle asynchronous data streams.

Tokens: 2.6K
Snippets: 5
Records: 18
Agent score: 11%

What's inside ro

Overview of samber/ro plugins
The samber/ro library provides a wide array of plugins to extend ReactiveX observables with specific domain functionality. Available plugins include:
- Data Manipulation: bytes (byte slices), strings (case conversions, locale-aware), strconv (conversions), sort (sorting/reordering), time (time manipulation), and exp/simd (SIMD-accelerated math).
- Encoding & Serialization: encoding/json, encoding/csv, encoding/base64, and encoding/gob.
- Scheduling & Timing: cron (cron expressions/intervals) and ICS (iCal calendars).
- Network & I/O: http/client (HTTP requests), io (file/stream I/O), fsnotify (file system monitoring), and websocket/client.
- Observability & Logging: observability/log, observability/zap, observability/logrus, observability/slog, observability/zerolog, and observability/sentry.
- Rate Limiting: ratelimit/native and ratelimit/ulule.
- Text Processing: regexp and template.
- System Integration: proc (process execution), signal (signal handling), and iter (iterators).
- Data Validation: ozzo/ozzo-validation.
- Utilities: hyperloglog (cardinality estimation), samber/hot (in-memory cache), samber/psi (starvation notifier), and testify.
https://ro.samber.dev/llms.txtwebsitellms
Understand ro core concepts
The ro library is built around four primary concepts:
- Observable: A stream of data that emits values over time.
- Observer: The consumer that handles Next, Error, and Complete events from an Observable.
- Operator: Functions used to transform, filter, or combine Observables.
- Subscription: The execution of an Observable, which can be cancelled.
https://ro.samber.dev/llms.txtwebsitellms

Get started with ro reactive programming in Go

The ro library provides a declarative way to handle asynchronous data streams in Go using the ReactiveX spec. You can create a stream using ro.Pipe, apply operators to transform or filter the data, and then consume the values by calling .Subscribe() with a handler like ro.OnNext().

import "github.com/samber/ro"
import "fmt"

// Basic reactive stream
observable := ro.Pipe(
    ro.Range(0, 10),
    ro.Filter(func(x int) bool {
        return x%2 == 0
    }),
    ro.Map(func(x int) string {
        return fmt.Sprintf("even-%d", x)
    }),
)

// Subscribe to consume values
observable.Subscribe(ro.OnNext(func(s string) {
    fmt.Println(s)
}))

// Output: even-0, even-2, even-4, even-6, even-8

Handle errors in reactive streams
Use these operators to manage failures and retries:
- Recovery: Catch (fallback Observable), OnErrorResumeNextWith (fallback Observables), OnErrorReturn (fallback value)
- Retries: Retry (infinite), RetryWithConfig (configurable options)
- Validation: ThrowIfEmpty (error if no items emitted)
https://ro.samber.dev/llms.txtwebsitellms

Perform asynchronous HTTP requests using ro-http-client

Use the github.com/samber/ro/plugin/http/client plugin to integrate HTTP requests into a reactive pipeline. Combine ro.FromSlice with ro.MergeMap to fetch multiple URLs concurrently.

import rohttp "github.com/samber/ro/plugin/http/client"

urls := []string{
    "https://api.github.com/users/samber",
    "https://api.github.com/users/ansible",
}

obs := ro.Pipe(
    ro.FromSlice(urls),
    ro.MergeMap(func(url string) ro.Observable[[]byte] {
        return rohttp.Get[[]byte](url)
    }),
    ro.Map(func(body []byte) map[string]any {
        var user map[string]any
        json.Unmarshal(body, &user)
        return user
    }),
    ro.Map(func(user map[string]any) string {
        return user["login"].(string)
    }),
)

obs.Subscribe(ro.OnNext(func(login string) {
    fmt.Printf("User: %s\n", login)
}))

Apply rate limiting to a stream with ro-ratelimit

Use the github.com/samber/ro/plugin/ratelimit/native plugin to limit the number of items processed per second in a pipeline.

import rorate "github.com/samber/ro/plugin/ratelimit/native"

obs := ro.Pipe(
    ro.Range(0, 100),
    rorate.Limit(10), // max 10 items per second
    ro.Map(func(i int) string {
        return fmt.Sprintf("Processing item %d", i)
    }),
)

obs.Subscribe(ro.OnNext(func(msg string) {
    fmt.Println(msg)
}))

Monitor file system events with ro-fsnotify

The github.com/samber/ro/plugin/fsnotify plugin allows you to create an observable that emits events when files are modified.

import rofsnotify "github.com/samber/ro/plugin/fsnotify"

obs := rofsnotify.Watch("./config.yaml")

obs.Subscribe(
    ro.OnNext(func(event fsnotify.Event) {
        fmt.Printf("File event: %s\n", event.Op.String())
    }),
    ro.OnError(func(err error) {
        fmt.Printf("Error: %v\n", err)
    }),
)

Validate data streams using ro-ozzo-validation

Integrate ozzo-validation into a reactive pipeline using the github.com/samber/ro/plugin/ozzo/ozzo-validation plugin to filter or handle invalid data objects.

import rovalidation "github.com/samber/ro/plugin/ozzo/ozzo-validation"

type User struct {
    Name  string `json:"name"`
    Email string `json:"email"`
    Age   int    `json:"age"`
}

obs := ro.Pipe(
    ro.FromChannel(userChannel),
    rovalidation.Validate(func(user User) error {
        return validation.ValidateStruct(&user,
            validation.Field(&user.Name, validation.Required),
            validation.Field(&user.Email, validation.Required, is.Email),
            validation.Field(&user.Age, validation.Required, validation.Min(18)),
        )
    }),
    ro.Map(func(user User) string {
        return fmt.Sprintf("Valid user: %s (%s)", user.Name, user.Email)
    }),
)

obs.Subscribe(ro.OnNext(func(msg string) {
    fmt.Println(msg)
}))

Convert Observables to Go types (Sink Operators)
Use these operators to extract data from a stream into standard Go structures:
- Non-blocking/Async: ToSlice (collect to slice), ToMap (collect to map), ToChannel (forward to channel)
- Blocking: Collect (collect all items into a slice), CollectWithContext (collect with context propagation)
https://ro.samber.dev/llms.txtwebsitellms
Filter data streams with Filtering Operators
Use these operators to control which items are emitted:
- Predicates: Filter (pass test), TakeWhile (while true), SkipWhile (while true), First (first matching), Last (last matching)
- Quantity: Take (first n), Skip (first n), TakeLast (last n), SkipLast (suppress last n), ElementAt (nth item), ElementAtOrDefault
- Uniqueness: Distinct (suppress duplicates), DistinctBy (duplicates based on key)
- Boundaries: TakeUntil (until signal emits), SkipUntil (until signal emits)
- Position: Head (first item), Tail (last item)
- Noise: IgnoreElements (only termination notifications)
https://ro.samber.dev/llms.txtwebsitellms
Apply conditional tests to Observables
Use these operators to evaluate the stream as a whole:
- Tests: All (all satisfy condition), Contains (any satisfy condition), SequenceEqual (compare two Observables)
- Selection: Find (first matching), DefaultIfEmpty (fallback value), Iif (return Observable based on condition)
https://ro.samber.dev/llms.txtwebsitellms
Combine multiple Observables
Use these operators to merge or synchronize multiple data streams:
- Merging: Merge (static or dynamic), MergeWith1-5, MergeAll, MergeMap (map then merge)
- Concatenation: Concat (sequential), ConcatWith, ConcatAll
- Zipping: Zip (in order), Zip2-6, ZipWith1-5, ZipAll
- Latest Values: CombineLatest2-5, CombineLatestAny, CombineLatestWith1-4, CombineLatestAll, CombineLatestAllAny
- Racing: Race (first to emit), Amb (alias for Race), RaceWith
- Padding: StartWith (values before source), EndWith (values after source)
- Sequencing: Pairwise (pairs of consecutive values)
https://ro.samber.dev/llms.txtwebsitellms