Programación reactiva en Go: librería RxGo (Reactive Extensions for Go)

Programación reactiva en Go (Golang) con la librería RxGo

La programación reactiva es un paradigma orientado a flujos de datos asíncronos y la propagación de cambios. En el ecosistema de Go (Golang), una de las implementaciones más potentes de este paradigma es la librería RxGo (Reactive Extensions for Go). Permite a los desarrolladores trabajar con secuencias de eventos utilizando un estilo declarativo, lo que es especialmente útil al procesar entrada de usuario, solicitudes de red, mensajes de canales y cualquier otra fuente de datos asíncrona.

¿Qué es RxGo?

RxGo es un puerto de la popular librería ReactiveX (Reactive Extensions) al lenguaje Go. La idea principal es crear un Observable (flujo observable) que emite elementos, y un Observer (observador) que se suscribe a este flujo y reacciona a los datos entrantes, errores o la finalización del flujo.

La librería proporciona un rico conjunto de operadores para transformar, filtrar, combinar y gestionar flujos. La ventaja clave es que el código se vuelve conciso, legible y fácil de probar, especialmente en escenarios complejos con múltiples fuentes de eventos.

Instalación de RxGo

Para empezar, instale el paquete mediante go get:

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

Asegúrese de usar la versión v2, ya que contiene la API actualizada y soporte para contextos de Go.

Conceptos básicos y ejemplos

1. Creación de un Observable

La forma más sencilla de crear un flujo es usar rxgo.Just o rxgo.FromChannel. Veamos un ejemplo con 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("Recibido:", item.V)        }    }}

Este código crea un flujo de cinco números. El método Observe() devuelve un canal por el que llegan los elementos. Importante: en RxGo se utiliza backpressure (contrapresión) — si el consumidor es lento, los datos se almacenan en búfer.

2. Operadores de transformación

Los operadores son el corazón de la programación reactiva. Por ejemplo, Map transforma cada elemento:

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)) // procesamiento paralelo

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

Salida: 10, 20, 30. El operador Map aplica una función a cada elemento. El parámetro rxgo.WithPool(2) habilita el procesamiento paralelo con dos gorutinas.

3. Filtrado y combinación de flujos

Con Filter se pueden descartar elementos no deseados, y Merge combina varios flujos:

package main

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

func main() {    // Filtrado de números pares    even := rxgo.Just(1, 2, 3, 4, 5)().        Filter(func(i interface{}) bool {            return i.(int)%2 == 0        })

    // Segundo flujo    odd := rxgo.Just(10, 20, 30)()

    // Combinación    merged := rxgo.Merge(even, odd)

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

4. Trabajo con canales y contextos

RxGo se integra perfectamente con los canales nativos de Go. Creemos un Observable a partir de un canal y agreguemos