Reactive eXtensions for Go
This is a package and tool providing Reactive eXtensions for Go.
The main package provides Rx operators for common Go builtin types, and the tool generates Go code for arbitrary types.
Why?
Yes, good question. Mostly as an exercise to see if it was feasible/possible. It is, largely, except for operators that produce sequence types (ie. arrays or observables of T).
That said, the Rx operators do provide some nice functionality that would otherwise have to be implemented by hand.
Installation
go get github.com/alecthomas/gorx github.com/alecthomas/gorx/cmd/gorx
Usage
To use the package:
import "github.com/alecthomas/gorx"
gorx.
FromTimeChannel(time.Tick(time.Second)).
Take(5).
Do(func(t time.Time) { fmt.Printf("%s\n", t) }).
Wait()
To generate Rx operators for custom types:
gorx --import=gopkg.in/alecthomas/kingpin.v2 kingpinrx '*kingpin.CmdClause' '*kingpin.FlagClause'
Examples
A very basic example creating an observable from a set of strings and printing them:
gorx.FromStrings("Ben", "George").Do(func(s string) { fmt.Println(s) }).Wait()
A more complex example. Try retrieving article from cache, otherwise fetch original from Wikipedia, all with a timeout.
func GetCached(url string) *ResponseStream {
fmt.Printf("No cache entry for %s\n", url)
return ThrowResponse(errors.New("not implemented"))
}
func SetCached(response *http.Response) {
fmt.Printf("Caching %s\n", response.Request.URL)
}
func Get(url string) *ResponseStream {
return StartResponse(func() (*http.Response, error) {
response, err := http.Get(url)
if err == nil && (response.StatusCode < 200 || response.StatusCode > 299) {
return nil, errors.New(http.StatusText(response.StatusCode))
}
return response, err
})
}
func URLForArticle(article string) string {
return "http://en.wikipedia.org/wiki/" + article
}
func LogError(err error) {
fmt.Printf("error: %s\n", err)
}
func GetWikipediaArticles(timeout time.Duration, articles ...string) *ResponseStream {
// Try cached URL first, then recover with remote URL and
// finally recover with an empty stream.
return FromStringArray(articles).
Map(URLForArticle).
FlatMapResponse(func(url string) ResponseObservable {
remote := Get(url).
Timeout(timeout).
Do(SetCached).
DoOnError(LogError).
Catch(EmptyResponse())
return GetCached(url).
Catch(remote)
})
}
Operators
Following are a list of the core operators as defined by reactivex.io that have been implemented and will (probably) be implemented soon:
Create operators
- Create
- Empty
- Never
- Throw
- Just
- Range
- Repeat
- Start
Not implemented:
- Defer
- Timer
Transformations
- Map
- Reduce
- Scan
- FlatMap
Not implemented:
- Buffer
- GroupBy
- Window
Note: These operators are currently not implemented because each distinct
observable type requires quite a lot of boilerplate code, and these operators
produce new types. eg. .Buffer(2)
would transform T
to a stream of []T
,
.FlatMap(f)
would transform T
to a TStreamStream
, etc. One "solution" is
to only generate these operators if the user explicitly requests these
resultant types.
Filters
- Distinct
- ElementAt
- Filter
- First
- Last
- Skip
- SkipLast
- Take
- TakeLast
- IgnoreElements
- Sample
- Debounce
Combining
- Merge
- MergeDelayError
Not implemented:
- CombineLatest
- And / Then / When
- Zip
- Join
- StartWith
- Switch
- Zip
Note: See note above for Transformations for why these are not implemented.
Error handling
- Catch
- Retry
Mathematics and Aggregation
- Concat
- Average
- Count
- Min
- Max
- Reduce
- Sum
Utility
- Do
- Subscribe
Not implemented:
- Delay
- Timeout
- Timestamp
- Materialize / Dematerialize
- Serialize
- TimeInterval
Conditional and Boolean
Not implemented:
- All
- Amb
- Contains
- DefaultIfEmpty
- SequenceEqual
- SkipUntil
- SkipWhile
- TakeUntil
- TakeWhile
Conversion
- To (one, array, channel)