add docs

README.md

@@ -7,7 +7,7 @@
 [![GitHub Release](https://img.shields.io/github/release-pre/jjeffcaii/reactor-go.svg)](https://github.com/jjeffcaii/reactor-go/releases)
 
 > A golang implementation for [reactive-streams](https://www.reactive-streams.org/).
-<br>🚧🚧🚧 ***IT IS UNDER ACTIVE DEVELOPMENT!!!*** 
+<br>🚧🚧🚧 ***IT IS UNDER ACTIVE DEVELOPMENT!!!***
 <br>***[WARNING] DO NOT USE IN ANY PRODUCTION ENVIRONMENT!!!***
 
 ## Install
@@ -108,4 +108,3 @@ func Example() {
 // next: #HELLO_0006
 // next: #HELLO_0008
 ```
-

flux/flux_create_sink.go

@@ -6,6 +6,7 @@ import (
 	"sync/atomic"
 
 	rs "github.com/jjeffcaii/reactor-go"
+	"github.com/jjeffcaii/reactor-go/hooks"
 )
 
 type bufferedSink struct {
@@ -39,11 +40,19 @@ func (p *bufferedSink) Complete() {
 }
 
 func (p *bufferedSink) Error(err error) {
+	if p.done {
+		hooks.Global().OnErrorDrop(err)
+		return
+	}
 	p.s.OnError(err)
 	p.dispose()
 }
 
 func (p *bufferedSink) Next(v interface{}) {
+	if p.done {
+		hooks.Global().OnNextDrop(v)
+		return
+	}
 	p.q.offer(v)
 	p.drain()
 }

flux/flux_error.go

@@ -0,0 +1,22 @@
+package flux
+
+import (
+	"context"
+
+	"github.com/jjeffcaii/reactor-go"
+	"github.com/jjeffcaii/reactor-go/internal"
+)
+
+type fluxError struct {
+	e error
+}
+
+func (p fluxError) SubscribeWith(ctx context.Context, s rs.Subscriber) {
+	actual := internal.NewCoreSubscriber(ctx, s)
+	actual.OnSubscribe(internal.EmptySubscription)
+	actual.OnError(p.e)
+}
+
+func newFluxError(e error) fluxError {
+	return fluxError{e: e}
+}

flux/flux_test.go

@@ -2,6 +2,7 @@ package flux_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"testing"
@@ -253,7 +254,6 @@ func TestEmpty(t *testing.T) {
 }
 
 func TestCreateWithRequest(t *testing.T) {
-
 	const totals = 20
 	f := flux.Create(func(ctx context.Context, sink flux.Sink) {
 		for i := 0; i < totals; i++ {
@@ -284,3 +284,33 @@ func TestCreateWithRequest(t *testing.T) {
 	f.SubscribeWith(context.Background(), sub)
 	assert.Equal(t, totals, int(processed), "bad processed num")
 }
+
+func TestError(t *testing.T) {
+	mockErr := errors.New("this is a mock error")
+	var sig rs.SignalType
+	var e1, e2 error
+	var requested int
+	flux.Error(mockErr).
+		DoFinally(func(s rs.SignalType) {
+			sig = s
+		}).
+		DoOnRequest(func(n int) {
+			requested = n
+		}).
+		DoOnError(func(e error) {
+			e1 = e
+		}).
+		Subscribe(
+			context.Background(),
+			rs.OnNext(func(v interface{}) {
+				assert.Fail(t, "should never run here")
+			}),
+			rs.OnError(func(e error) {
+				e2 = e
+			}),
+		)
+	assert.Equal(t, mockErr, e1, "bad doOnError")
+	assert.Equal(t, mockErr, e2, "bad onError")
+	assert.Equal(t, rs.SignalTypeError, sig, "bad signal")
+	assert.True(t, requested > 0, "no request")
+}

flux/misc.go

@@ -1,6 +1,8 @@
 package flux
 
-import "errors"
+import (
+	"errors"
+)
 
 const (
 	statCancel   = -1

flux/processor_test.go

@@ -41,21 +41,3 @@ func TestUnicastProcessor(t *testing.T) {
 	}))
 	<-done
 }
-
-func TestDoOnSubscribe(t *testing.T) {
-	pc := flux.NewUnicastProcessor()
-	time.AfterFunc(1*time.Second, func() {
-		pc.Next(111)
-		pc.Next(222)
-		pc.Complete()
-	})
-	done := make(chan struct{})
-	pc.DoFinally(func(s rs.SignalType) {
-		close(done)
-	}).DoOnSubscribe(func(su rs.Subscription) {
-		log.Println("doOnSubscribe")
-	}).BlockLast(context.Background())
-
-	<-done
-
-}

flux/queue.go

@@ -1,32 +1,23 @@
 package flux
 
-import "io"
-
-type queue interface {
-	io.Closer
-	offer(interface{})
-	poll() (interface{}, bool)
-	size() int
-}
-
-type simpleQueue struct {
+type queue struct {
 	c chan interface{}
 }
 
-func (q simpleQueue) size() int {
+func (q queue) size() int {
 	return len(q.c)
 }
 
-func (q simpleQueue) Close() (err error) {
+func (q queue) Close() (err error) {
 	close(q.c)
 	return
 }
 
-func (q simpleQueue) offer(v interface{}) {
+func (q queue) offer(v interface{}) {
 	q.c <- v
 }
 
-func (q simpleQueue) poll() (v interface{}, ok bool) {
+func (q queue) poll() (v interface{}, ok bool) {
 	select {
 	case v, ok = <-q.c:
 		return
@@ -36,7 +27,7 @@ func (q simpleQueue) poll() (v interface{}, ok bool) {
 }
 
 func newQueue(size int) queue {
-	return simpleQueue{
+	return queue{
 		c: make(chan interface{}, size),
 	}
 }

flux/utils.go

@@ -9,10 +9,7 @@ import (
 var empty = just(nil)
 
 func Error(e error) Flux {
-	// TODO: need implementation
-	return Create(func(ctx context.Context, sink Sink) {
-		sink.Error(e)
-	})
+	return wrap(newFluxError(e))
 }
 
 func Range(start, count int) Flux {

internal/misc.go

@@ -3,9 +3,14 @@ package internal
 import (
 	"errors"
 	"fmt"
+
+	rs "github.com/jjeffcaii/reactor-go"
 )
 
-var ErrCallOnSubscribeDuplicated = errors.New("call OnSubscribe duplicated")
+var (
+	ErrCallOnSubscribeDuplicated                 = errors.New("call OnSubscribe duplicated")
+	EmptySubscription            rs.Subscription = &emptySubscription{}
+)
 
 func TryRecoverError(re interface{}) error {
 	if re == nil {
@@ -20,3 +25,12 @@ func TryRecoverError(re interface{}) error {
 		return fmt.Errorf("%s", e)
 	}
 }
+
+type emptySubscription struct {
+}
+
+func (emptySubscription) Request(n int) {
+}
+
+func (emptySubscription) Cancel() {
+}

mono/mono_error.go

@@ -0,0 +1,22 @@
+package mono
+
+import (
+	"context"
+
+	"github.com/jjeffcaii/reactor-go"
+	"github.com/jjeffcaii/reactor-go/internal"
+)
+
+type monoError struct {
+	e error
+}
+
+func (p monoError) SubscribeWith(ctx context.Context, s rs.Subscriber) {
+	actual := internal.NewCoreSubscriber(ctx, s)
+	actual.OnSubscribe(internal.EmptySubscription)
+	actual.OnError(p.e)
+}
+
+func newMonoError(e error) monoError {
+	return monoError{e: e}
+}

mono/mono_test.go

@@ -306,3 +306,28 @@ func BenchmarkCreate(b *testing.B) {
 		}
 	})
 }
+
+func TestError(t *testing.T) {
+	mockErr := errors.New("this is a mock error")
+	var sig rs.SignalType
+	var e1, e2 error
+	mono.Error(mockErr).
+		DoFinally(func(s rs.SignalType) {
+			sig = s
+		}).
+		DoOnError(func(e error) {
+			e1 = e
+		}).
+		Subscribe(
+			context.Background(),
+			rs.OnNext(func(v interface{}) {
+				assert.Fail(t, "should never run here")
+			}),
+			rs.OnError(func(e error) {
+				e2 = e
+			}),
+		)
+	assert.Equal(t, mockErr, e1, "bad doOnError")
+	assert.Equal(t, mockErr, e2, "bad onError")
+	assert.Equal(t, rs.SignalTypeError, sig, "bad signal")
+}

mono/processor_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+
 func TestProcessor(t *testing.T) {
 	p := mono.CreateProcessor()
 

mono/utils.go

@@ -11,6 +11,10 @@ import (
 var empty = wrap(newMonoJust(nil))
 var errJustNilValue = errors.New("require non nil value")
 
+func Error(e error) Mono {
+	return wrap(newMonoError(e))
+}
+
 func Empty() Mono {
 	return empty
 }
@@ -29,7 +33,7 @@ func Just(v interface{}) Mono {
 	return wrap(newMonoJust(v))
 }
 
-func Create(gen func(context.Context, Sink)) Mono {
+func Create(gen func(ctx context.Context, s Sink)) Mono {
 	return wrap(newMonoCreate(gen))
 }
 

reactor.go

@@ -25,8 +25,8 @@ const (
 )
 
 type (
-	Predicate func(interface{}) bool
-	Transformer func(interface{}) interface{}
+	Predicate func(v interface{}) bool
+	Transformer func(v interface{}) interface{}
 
 	FnOnComplete = func()
 	FnOnNext = func(v interface{})