RxGo: Reactive Extensions for the Go

RxGo is a Go implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.

V0.2

function program friendly
Connectable observables, support Hot or Cold item emiting
context support for concurrency
support error type data in observable sequences
internal debug support
unsubscribe supporting
non-opinionated about source of concurrency
async or synchronous execution

Getting started

Hello World

The Hello World program:

package main

import (
	"fmt"
	RxGo "github.com/pmlpml/rxgo"
)

func main() {
	RxGo.Just("Hello", "World", "!").Subscribe(func(x string) {
		fmt.Println(x)
	})
}

output:

Hello
World
!

Chained operations on stream

The dataflows in RxGo consist of a source, zero or more intermediate steps followed by a data consumer or combinator step (where the step is responsible to consume the dataflow by some means):

source.operator1().operator2().operator3().subscribe(observer);

The first obserable emits items, so called source. And a chained operations apply on each item asynchronously, until subscribe to the observer.

The sample program generate a fibonacci sequence, and then double each item.

package main

import (
	"fmt"
	RxGo "github.com/pmlpml/rxgo"
)

func fibonacci(max int) func() (int, bool) {
	a, b := 0, 1
	return func() (r int, end bool) {
		r = a
		a, b = b, a+b
		if r > max {
			end = true
		}
		return
	}
}

func main() {
	RxGo.Start(fibonacci(10)).Map(func(x int) int {
		return 2*x
	}).Subscribe(func(x int) {
		fmt.Print(x)
	})
}

The result is 022461016. the source Start(fibonacci(10)) generates dataflow 0112358 to a operation Map, and Subscribe to print

Connectable observables

A Connectable Observable resembles an ordinary Observable, except that it does not begin emitting items when it is subscribed to, but only when its connect() method is called. There are two phases for data pipeline.

The first phase is called definition. we define source function or operators for a pipeline that is the same as assign work roles for each node on it

The next phase is called runtime. when any observable call Subscribe(...) , the pipeline include this observable will connect its all node or appoint one or more workers stand aside by each node. if predecess worker emit a data, the worker will play the role defined before, and send result to the next worker

The simple program shows the function of a Pipeline Restart

package main

import (
	"fmt"
	RxGo "github.com/pmlpml/rxgo"
)

func main() {
    //define pipeline
	source := RxGo.Just("Hello", "World", "!")
	next := source.Map(func(s string) string {
		return s + " "
	})
    //run pipeline
	next.Subscribe(func(x string) {
		fmt.Print(x)
	})
	fmt.Println()
	source.Subscribe(func(x string) {
		fmt.Print(x)
	})
}