rs

README

reactor-go 🚀🚀🚀

A golang implementation for reactive-streams. [WARNNING] IT IS UNDER ACTIVE DEVELOPMENT!!! DO NOT USE IN ANY PRODUCTION ENVIRONMENT!!!

🏠 Homepage

Install

go get -u github.com/jjeffcaii/reactor-go

Example

NOTICE:
We can only use func(interface{})interface{} for most operations because Golang has not Generics. 😭
If you have any better idea, please let me know. 😀

Mono

package mono_test

import (
	"context"
	"fmt"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/mono"
)

func Example() {
	gen := func(ctx context.Context, sink mono.Sink) {
		sink.Success("World")
	}
	mono.
		Create(gen).
		Map(func(i interface{}) interface{} {
			return "Hello " + i.(string) + "!"
		}).
		DoOnNext(func(s rs.Subscription, v interface{}) {
			fmt.Println(v)
		}).
		Subscribe(context.Background())
}

// Should print
// Hello World!

Flux

package rs_test

import (
	"context"
	"fmt"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/jjeffcaii/reactor-go/scheduler"
)

func Example() {
	gen := func(ctx context.Context, sink flux.Sink) {
		for i := 0; i < 10; i++ {
			v := i
			sink.Next(v)
		}
		sink.Complete()
	}
	done := make(chan struct{})
	flux.Create(gen).
		Filter(func(i interface{}) bool {
			return i.(int)%2 == 0
		}).
		Map(func(i interface{}) interface{} {
			return fmt.Sprintf("#HELLO_%04d", i.(int))
		}).
		SubscribeOn(scheduler.Elastic()).
		Subscribe(context.Background(),
			rs.OnSubscribe(func(s rs.Subscription) {
				s.Request(1)
			}),
			rs.OnNext(func(s rs.Subscription, v interface{}) {
				fmt.Println("next:", v)
				s.Request(1)
			}),
			rs.OnComplete(func() {
				close(done)
			}),
		)
	<-done
}
// Should print:
// next: #HELLO_0000
// next: #HELLO_0002
// next: #HELLO_0004
// next: #HELLO_0006
// next: #HELLO_0008

Author

👤 Jeffsky

• Github: @jjeffcaii

Show your support

Give a ⭐️ if this project helped you!

---

This README was generated with ❤️ by readme-md-generator

Documentation

Index

• Constants
• Variables
• type Disposable
• type FnOnCancel
• type FnOnComplete
• type FnOnError
• type FnOnFinally
• type FnOnNext
• type FnOnRequest
• type FnOnSubscribe
• type Predicate
• type Processor
• type Publisher
• type RawPublisher
• type Signal
  • func (s Signal) String() string
• type Subscriber
  • func NewSubscriber(opts ...SubscriberOption) Subscriber
• type SubscriberOption
  • func OnComplete(onComplete FnOnComplete) SubscriberOption
  • func OnError(onError FnOnError) SubscriberOption
  • func OnNext(onNext FnOnNext) SubscriberOption
  • func OnSubscribe(onSubscribe FnOnSubscribe) SubscriberOption
• type Subscription
• type Transformer

Constants

const RequestInfinite = math.MaxInt32

Variables

var ErrCancelled = errors.New("subscriber has been cancelled")

Types

type Disposable

type Disposable interface {
	// Dispose dispose current resource.
	Dispose()
	// IsDisposed returns true if it has been disposed.
	IsDisposed() bool
}

Disposable is a disposable resource.

type FnOnCancel

type FnOnCancel = func()

type FnOnComplete

type FnOnComplete = func()

type FnOnError

type FnOnError = func(error)

type FnOnFinally

type FnOnFinally = func(Signal)

type FnOnNext

type FnOnNext = func(s Subscription, v interface{})

type FnOnRequest

type FnOnRequest = func(int)

type FnOnSubscribe

type FnOnSubscribe = func(Subscription)

type Predicate

type Predicate func(interface{}) bool

type Processor

type Processor interface {
	Publisher
	Subscriber
}

type Publisher

type Publisher interface {
	RawPublisher
	Subscribe(context.Context, ...SubscriberOption)
}

type RawPublisher

type RawPublisher interface {
	SubscribeWith(context.Context, Subscriber)
}

type Signal

type Signal int8

const (
	SignalDefault Signal = iota
	SignalComplete
	SignalCancel
	SignalError
)

func (Signal) String

func (s Signal) String() string

type Subscriber

type Subscriber interface {
	OnComplete()
	OnError(error)
	OnNext(Subscription, interface{})
	OnSubscribe(Subscription)
}

func NewSubscriber

func NewSubscriber(opts ...SubscriberOption) Subscriber

type SubscriberOption

type SubscriberOption func(*subscriber)

func OnComplete

func OnComplete(onComplete FnOnComplete) SubscriberOption

func OnError

func OnError(onError FnOnError) SubscriberOption

func OnNext

func OnNext(onNext FnOnNext) SubscriberOption

func OnSubscribe

func OnSubscribe(onSubscribe FnOnSubscribe) SubscriberOption

type Subscription

type Subscription interface {
	Request(n int)
	Cancel()
}

type Transformer

type Transformer func(interface{}) interface{}