Reactive Programming in Go (Golang): Libraries, Patterns, and Examples

Reactive Programming in Go (Golang)

Reactive programming is a paradigm focused on asynchronous data streams and the propagation of changes. In the Go language (Golang), this concept finds a natural implementation thanks to built-in goroutines and channels. However, for complex scenarios, there are specialized libraries that provide powerful tools for working with reactive streams.

In this article, we will look at the main libraries for reactive programming in Go, their capabilities, and usage examples. The material will be useful for both beginners and experienced developers who want to implement reactive approaches in their projects.

Basic Concepts of Reactive Programming

Before moving on to the tools, it is important to understand the key terms:

• Observable — a data source that can generate events.
• Observer — a subscriber that reacts to events.
• Operators — functions for transforming, filtering, and combining streams (map, filter, merge, flatMap).
• Backpressure — a mechanism for controlling the data transfer rate between producer and consumer.

RxGo Library (Reactive Extensions for Go)

RxGo is the most popular implementation of reactive extensions for Go. It ports concepts from RxJava/RxJS, adapting them to the language's features (goroutines, lack of inheritance).

Installation:

go get github.com/reactivex/rxgo/v2

Example: Basic Observable

package main

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

func main() {    // Create an Observable from a list    observable := rxgo.Just(1, 2, 3, 4, 5)()

    // Subscribe and print elements    ch := observable.Observe()    for item := range ch {        fmt.Println(item.V)    }}

Example: Transformation with Operators

package main

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

func main() {    observable := rxgo.Just(1, 2, 3, 4, 5)().        Map(func(item rxgo.Item) rxgo.Item {            val := item.V.(int)            return rxgo.Of(val * 2)        }).        Filter(func(item rxgo.Item) bool {            return item.V.(int) > 5        })

    for item := range observable.Observe() {        fmt.Println(item.V) // Output: 6, 8, 10    }}

go-reactive Library

go-reactive is a lightweight alternative to RxGo, focused on simplicity and performance. It uses Go channels as the basis for reactive streams.

Installation:

go get github.com/teivah/go-reactive

Example: Reactive Stream with Channels

package main

import (    "fmt"    "github.com/teivah/go-reactive")

func main() {    // Create a stream of numbers    stream := reactive.FromSlice([]int{10, 20, 30, 40, 50})

    // Apply operators    result := stream.        Filter(func(i int) bool { return i > 20 }).        Map(func(i int) int { return i * 3 })

    // Get the result    for v := range result.Out() {        fmt.Println(v) // Output: 90, 120, 150    }}

Reactive Programming with Native Channels

Go allows implementing reactive patterns without third-party libraries, using goroutines and channels. This is especially useful for simple scenarios.

Example: Observer Pattern with Channels

package main

import (    "fmt"    "time")

type Event struct {    Data string}

// Subject — the observable objectfunc Subject(out chan<- Event) {    for i := 0; i < 5; i++ {        out <- Event{Data: fmt.Sprintf("Event #%d", i)}        time.Sleep(100 * time.Millisecond)    }    close(out)}

// Observer — the observerfunc Ob