README
¶
Reactive 
My attempt on creating a simple RxJs clone
Features
- Observables
- Multi-Type support
- Subjects
- Subject
- ReplaySubject
- Pipes
- Take
- TakeEvery
- Skip
- SkipEvery
Examples
Simple Subject
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewSubject()
subject.Subscribe(subHandler)
subject.Next(1)
subject.Next(2)
subject.Next(3)
subject.Next(4)
}
func subHandler(a int) {
fmt.Println(a)
}
Output
$ go run main.go
1
2
3
4
Replay Subject
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewReplaySubject()
subject.Next(1)
subject.Next(2)
subject.Next(3)
subject.Subscribe(subHandler)
subject.Next(4)
}
func subHandler(a int) {
fmt.Println(a)
}
Output
$ go run main.go
3
4
Multi-Type support
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewSubject()
subject.Subscribe(intHandler)
subject.Subscribe(stringHandler)
subject.Next(2)
subject.Next("Hello")
subject.Next("World")
subject.Next(4)
subject.Next(nil)
}
func intHandler(a int) {
fmt.Print("Int Handler: ")
fmt.Println(a)
}
func stringHandler(a string) {
fmt.Print("String Handler: ")
fmt.Println(a)
}
Output
Int Handler: 2
String Handler: Hello
String Handler: World
Int Handler: 4
Int Handler: 0
String Handler:
Take Pipe
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewReplaySubject()
subject.Pipe(reactive.Take(2)).Subscribe(subHandler)
subject.Next(1)
subject.Next(2)
subject.Next(3)
subject.Next(4)
}
func subHandler(a int) {
fmt.Println(a)
}
Output
$ go run main.go
1
2
TakeEvery Pipe
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewReplaySubject()
subject.Pipe(reactive.TakeEvery(2)).Subscribe(subHandler)
subject.Next(1)
subject.Next(2)
subject.Next(3)
subject.Next(4)
}
func subHandler(a int) {
fmt.Println(a)
}
Output
$ go run main.go
2
4
Skip Pipe
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewReplaySubject()
subject.Pipe(reactive.Skip(2)).Subscribe(subHandler)
subject.Next(1)
subject.Next(2)
subject.Next(3)
subject.Next(4)
}
func subHandler(a int) {
fmt.Println(a)
}
Output
$ go run main.go
3
4
SkipEvery Pipe
package main
import (
"github.com/infinytum/reactive"
"fmt"
)
func main() {
subject := reactive.NewReplaySubject()
subject.Pipe(reactive.SkipEvery(2)).Subscribe(subHandler)
subject.Next(1)
subject.Next(2)
subject.Next(3)
subject.Next(4)
}
func subHandler(a int) {
fmt.Println(a)
}
Output
$ go run main.go
1
3
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Skip ¶
func Skip(count int) func(Observable, Subjectable)
Skip will ignore a specified amount of updates and will pass through all following
func SkipEvery ¶
func SkipEvery(count int) func(Observable, Subjectable)
SkipEvery will skip every {count} update and will pass all others
func Take ¶
func Take(count int) func(Observable, Subjectable)
Take automatically unsubscribes an observable after the given amount of times it has been updated
func TakeEvery ¶
func TakeEvery(count int) func(Observable, Subjectable)
TakeEvery only passes every {count} update to the registered function
Types ¶
type Observable ¶
type Observable interface {
// AsChannel returns a channel which will receive all
// further updates of this observable
AsChannel() chan []interface{}
// Pipe decorates an observable with one or multiple middlewares
// and returns a new observable with the decoration applied
Pipe(fns ...func(Observable, Subjectable)) Observable
// Subscribe registers a function for further updates of
// this observable and returns a subscription token which can
// be used to unsubscribe from it at any time
Subscribe(fn interface{}) (Subscription, error)
// Unsubscribe unregisters a previously registered function for all
// further updates of this observable or until re-registering.
Unsubscribe(subscription Subscription) error
}
Observable defines the requirements for a class to be considered a valid observable
type Pipe ¶
type Pipe func(Observable, Subjectable)
Pipe is the method signature for a pipe function
type Subjectable ¶
type Subjectable interface {
Observable
// Close will remove all subscribers and render
// the subjectable useless
Close()
// Next takes an undefined amount of parameters
// which will be passed to subscribed functions
Next(values ...interface{})
}
Subjectable defines required methods for an object to be considered a subject
func NewBufferSubject ¶
func NewBufferSubject(bufferSize int) Subjectable
NewBufferSubject returns a pointer to an empty instance of bufferSubject
func NewReplaySubject ¶
func NewReplaySubject() Subjectable
NewReplaySubject returns a pointer to an empty instance of replaySubject
func NewSubject ¶
func NewSubject() Subjectable
NewSubject returns a pointer to an empty instance of subject
type Subscription ¶
Subscription represents a string to identify a subscription in an obserable so it can be removed
func EmptySubscription ¶
func EmptySubscription() Subscription
func NewSubscription ¶
func NewSubscription() Subscription
NewSubscription generates a new subscription
Source Files
¶
Directories
Click to show internal directories.
Click to hide internal directories.