Reactive Programming in Go with samber/ro
Created:
目次
Reactive Programming with samber/ro
Introduction
This post provides a brief introduction and personal insights on experimenting with reactive programming in Go using the samber/ro library.
For comprehensive documentation and full feature references, please consult the official resources:
Basics
Observable
Everything begins with creating an Observable.
ro.Just(10, 20, 30, 40) // ro.Observable[int]Operators
Operations such as transformation and filtering are chained using Pipe functions:
pipeline := ro.Pipe1(
ro.Just(10, 20, 30, 40),
ro.Map(func(val int) int { return val * 2 }),
) The resulting pipeline is also of type ro.Observable[int]. Because only a single operator was used here, Pipe1 was applied, but Pipe2, Pipe3, Pipe4, etc., are available for longer pipelines.
Alternatively, Pipe[First, Last any] can be used without specifying the exact operator count:
pipeline := ro.Pipe[int, int](
ro.Just(10, 20, 30, 40),
ro.Map(func(val int) int { return val * 2 }),
)This functions identically to the previous snippet. It offers flexibility during prototyping because adding or removing operators does not require changing the function name. However, PipeN functions are preferred in production because type inference allows omitting explicit type parameters ([int, int]), which is also recommended in the official documentation.
Subscription
As is typical in reactive programming, constructing a pipeline does not trigger execution on its own; a subscriber must be attached.
pipeline.Subscribe(ro.OnNext(func(val int) {
fmt.Println(val) // 20, 40, 60, 80
}))You can handle onError and onCompleted events by supplying an observer via ro.NewObserver:
pipeline.Subscribe(ro.NewObserver(
func(val int) {
// onNext
fmt.Println(val)
},
func(err error) {
// onError
fmt.Printf("Stream error: %v\n", err)
},
func() {
// onCompleted
fmt.Println("All streams completed!")
},
))Alternatively, ro.Collect can be used to gather all emitted items into a slice (note that this is a blocking operation, but it is useful for debugging and testing):
result, err := ro.Collect(pipeline) // result is of type []intError Handling
When an operation inside the pipeline can return an error, MapErr must be used instead of Map. MapErr accepts a function with the signature func(T) (R, error).
obs := ro.Pipe[int, int](
ro.Just(1, 2, 3),
ro.MapErr(func(i int) (int, error) {
if i == 2 {
return 0, errors.New("number 2 is not allowed")
}
return i * 2, nil
}),
ro.Catch(func(err error) ro.Observable[int] {
fmt.Printf("Error: %v\n", err)
return ro.Just(99) // Fallback value
}),
)
result, err := ro.Collect(obs)
fmt.Println("Result:", result) // Result: [2 99]
fmt.Println("Error:", err) // Error: <nil>In this example, ro.Catch handles errors emitted by earlier operators. However, emitting a fallback value from Catch terminates the stream; subsequent items are not processed. As shown in the output (Result: [2 99]), the item 3 is never processed.
When I want the process to continue even if an error occurs, I’ve set the Map function to return a fallback value instead of throwing an error, as shown below.
obs := ro.Pipe[int, int](
ro.Just(1, 2, 3),
ro.Map(func(i int) int {
if i == 2 {
// Fallback value on error
return 99
}
return i * 2
}),
)Empty
You can also return ro.Empty to discard specific items. When an empty observable is returned, subsequent operators skip that item.
(When returning an Observable instead of a plain value, use FlatMap instead of Map)
obs := ro.Pipe[int, int](
ro.Just(1, 2, 3),
ro.FlatMap(func(i int) ro.Observable[int] {
if i == 2 {
fmt.Println("number 2 will be empty")
return ro.Empty[int]()
}
return ro.Just(i * 2)
}),
ro.Map(func(i int) int {
return i * 10
}),
)
result, err := ro.Collect(obs)
fmt.Println("Result:", result) // Result: [20 60]
fmt.Println("Error:", err) // Error: <nil>In the example above, 2 is skipped, and only 1 and 3 proceed through the pipeline. Returning ro.Empty on error can also be a viable pattern depending on requirements.
Passing Multiple Values Downstream
When chaining operations, you often need to forward both the original input and a computed value to subsequent stages. While declaring a dedicated struct is the standard approach in Go, creating temporary structs solely for passing values through a pipeline can feel cumbersome.
In such cases, Tuple from the samber/lo library (by the same author) can be helpful:
pipeline := ro.Pipe2(
ro.Just("apple", "banana", "cherry", "date"),
ro.Map(func(word string) lo.Tuple2[string, int] {
l := len(word) // Simulates an operation such as a DB query
return lo.T2(word, l)
}),
ro.Map(func(pair lo.Tuple2[string, int]) int {
fmt.Printf("word: %s, len: %d\n", pair.A, pair.B)
return pair.B
}),
)However, heavy reliance on tuples can obscure field semantics (pair.A, pair.B). It is best reserved for simple cases; dedicated structs are preferable for complex domain logic.
Testing
Consider testing a function that returns an ro.Observable:
func example() ro.Observable[int] {
pipeline := ro.Pipe[int, int](
ro.Just(10, 20, 30, 40),
ro.Map(func(val int) int { return val * 2 }),
)
return pipeline
}You can test this straightforwardly using ro.Collect:
func TestExample(t *testing.T) {
got, _ := ro.Collect(example())
want := []int{20, 40, 60, 80}
if !slices.Equal(got, want) {
t.Errorf("example() = %v, want %v", got, want)
}
}Additionally, ro provides a dedicated testing package with fluent assertions:
import rotesting "github.com/samber/ro/testing"
func TestExample(t *testing.T) {
rotesting.Assert[int](t).
Source(example()).
ExpectNext(20).
ExpectNext(40).
ExpectNext(60).
ExpectNext(80).
// ExpectNextSeq(20, 40, 60, 80). // Values can also be asserted in batch
ExpectComplete().
Verify()
}Key points:
ExpectComplete: Asserts that the stream completed successfully. Without this, unexpected trailing emissions (e.g., an unexpected100) will not trigger a test failure.Verify: Subscribes to the stream to initiate execution. Without callingVerify, the observable is not subscribed to and no assertions run.
This fluent style aligns well with reactive testing patterns. The testing package also provides additional capabilities for complex scenarios.
When Should It Be Used?
For the simple transformations demonstrated in this article, standard for loops are clearer and more idiomatic in Go; reactive programming offers little advantage in such scenarios.
The official documentation outlines the following use cases:
- Real-time data processing (WebSocket events, sensor data)
- User interface events (clicks, keystrokes, form inputs)
- API response handling (with retry, timeout, and caching)
- Data processing with transformation, aggregation and enrichment
- Event-driven patterns
While typical Go backend services may rarely require this paradigm, rich CLI or TUI applications represent a compelling candidate. Interactive tools like Claude Code or Codex must manage simultaneous user input, keystroke shortcuts, stream rendering, and asynchronous cancellation—fitting into the “User interface events” category.
Conversely, standard web applications—which primarily follow sequential steps such as parsing JSON, querying a database, and returning a response—derive little benefit from reactive streams. While ro supports operators for retries and timeouts, standard Go idioms are often simpler and more maintainable for sequential workflows.
Furthermore, Go’s syntax does not naturally lend itself to reactive chaining. Ideally, one would write fluent chains such as:
ro.Just(10, 20, 30, 40).Map(i => i*2).FlatMap(x => foo(x))However, Go lacks lambda expressions and multiple return values (e.g., (T, error)) do not fit neatly into single-argument chains. More fundamentally, Go’s type system does not support declaring new type parameters on methods, precluding fluent method chaining where types change across operations. Consequently, pipelines must be constructed using PipeN functions, which can become cumbersome when modifying operators later:
obs := ro.Pipe3(
source,
ro.Filter(predicate),
ro.Map(transformer),
ro.Take[int](10),
)