Programação Reativa em Go (Golang) com RxGo

Programação Reativa em Go (Golang) com a biblioteca RxGo

Programação reativa é um paradigma orientado a fluxos de dados assíncronos e propagação de mudanças. No ecossistema Go (Golang), uma das implementações mais poderosas desse paradigma é a biblioteca RxGo (Reactive Extensions for Go). Ela permite que desenvolvedores trabalhem com sequências de eventos usando um estilo declarativo, o que é especialmente útil ao processar entrada do usuário, requisições de rede, mensagens de canais e quaisquer outras fontes de dados assíncronas.

O que é RxGo?

RxGo é uma porta da popular biblioteca ReactiveX (Reactive Extensions) para a linguagem Go. A ideia principal é criar um Observable (fluxo observável), que emite elementos, e um Observer (observador), que se inscreve nesse fluxo e reage aos dados recebidos, erros ou conclusão do fluxo.

A biblioteca fornece um rico conjunto de operadores para transformação, filtragem, combinação e gerenciamento de fluxos. A principal vantagem é que o código se torna conciso, legível e facilmente testável, especialmente em cenários complexos com múltiplas fontes de eventos.

Instalação do RxGo

Para começar, instale o pacote via go get:

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

Certifique-se de usar a versão v2, pois ela contém a API atualizada e suporte para contextos Go.

Conceitos básicos e exemplos

1. Criando um Observable

A maneira mais simples de criar um fluxo é usar rxgo.Just ou rxgo.FromChannel. Veja um exemplo com 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("Erro:", item.Err)        } else {            fmt.Println("Recebido:", item.V)        }    }}

Este código cria um fluxo de cinco números. O método Observe() retorna um canal por onde os elementos chegam. Importante: no RxGo, é usado backpressure (contrapressão) — se o consumidor for lento, os dados são armazenados em buffer.

2. Operadores de transformação

Operadores são o coração da programação reativa. Por exemplo, 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)) // processamento paralelo

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

Saída: 10, 20, 30. O operador Map aplica uma função a cada elemento. O parâmetro rxgo.WithPool(2) ativa o processamento paralelo com duas goroutines.

3. Filtragem e combinação de fluxos

Com Filter você pode filtrar elementos indesejados, e Merge combina vários fluxos:

package main

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

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

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

    // Combinação    merged := rxgo.Merge(even, odd)

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