Reactive Programming in Go (Golang) with RxGo

Reactive Programming in Go (Golang) with the RxGo Library

Reactive programming is a paradigm focused on asynchronous data streams and the propagation of changes. In the Go (Golang) ecosystem, one of the most powerful implementations of this paradigm is the RxGo library (Reactive Extensions for Go). It allows developers to work with sequences of events using a declarative style, which is especially useful when processing user input, network requests, messages from channels, and any other asynchronous data sources.

What is RxGo?

RxGo is a port of the popular ReactiveX library (Reactive Extensions) to the Go language. The core idea is to create an Observable (an observable stream) that emits items, and an Observer that subscribes to this stream and reacts to incoming data, errors, or stream completion.

The library provides a rich set of operators for transforming, filtering, combining, and managing streams. The key advantage is that the code becomes concise, readable, and easily testable, especially in complex scenarios with multiple event sources.

Installing RxGo

To get started, install the package via go get:

go get -u github.com/reactivex/rxgo/v2

Make sure to use version v2, as it contains the current API and support for Go contexts.

Basic Concepts and Examples

1. Creating an Observable

The simplest way to create a stream is to use rxgo.Just or rxgo.FromChannel. Let's look at an example with Just:

package main

import (    "context"    "fmt"    "github.com/reactivex/rxgo/v2")

func main() {    observable := rxgo.Just(1, 2, 3, 4, 5)()    ch := observable.Observe()    for item := range ch {        if item.Error() {            fmt.Println("Error:", item.Err)        } else {            fmt.Println("Received:", item.V)        }    }}

This code creates a stream of five numbers. The Observe() method returns a channel through which items arrive. Important: RxGo uses backpressure — if the consumer is slow, data is buffered.

2. Transformation Operators

Operators are the heart of reactive programming. For example, Map transforms each item:

package main

import (    "context"    "fmt"    "github.com/reactivex/rxgo/v2")

func main() {    observable := rxgo.Just(1, 2, 3)().        Map(func(ctx context.Context, i interface{}) (interface{}, error) {            num := i.(int)            return num * 10, nil        }, rxgo.WithPool(2)) // parallel processing

    for item := range observable.Observe() {        fmt.Println(item.V)    }}

Output: 10, 20, 30. The Map operator applies a function to each item. The parameter rxgo.WithPool(2) enables parallel processing with two goroutines.

3. Filtering and Merging Streams

Using Filter you can discard unwanted items, and Merge combines multiple streams:

package main

import (    "fmt"    "github.com/reactivex/rxgo/v2")

func main() {    // Filtering even numbers    even := rxgo.Just(1, 2, 3, 4, 5)().        Filter(func(i interface{}) bool {            return i.(int)%2 == 0        })

    // Second stream    odd := rxgo.Just(10, 20, 30)()

    // Merging    merged := rxgo.Merge(even, odd)

    for item := range merged.Observe() {        fmt.Println(item.V) // 2, 4, 10, 20, 30    }}