fix(mono): add panic recover for most of Mono operations (#30)

* fix(mono): add panic recover for most of the Mono operations

* fix: lint

flux/op_filter.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/internal"
+	"github.com/pkg/errors"
 )
 
 type fluxFilter struct {
@@ -42,11 +43,23 @@ func (p *filterSubscriber) OnError(err error) {
 }
 
 func (p *filterSubscriber) OnNext(v Any) {
+	if p.f == nil {
+		p.OnError(errors.New("the Filter predicate is nil"))
+		return
+	}
+
 	defer func() {
-		if err := internal.TryRecoverError(recover()); err != nil {
-			p.OnError(err)
+		rec := recover()
+		if rec == nil {
+			return
+		}
+		if e, ok := rec.(error); ok {
+			p.OnError(errors.WithStack(e))
+		} else {
+			p.OnError(errors.Errorf("%v", rec))
 		}
 	}()
+
 	if p.f(v) {
 		p.actual.OnNext(v)
 		return

internal/misc.go

@@ -2,26 +2,11 @@ package internal
 
 import (
 	"errors"
-	"fmt"
 )
 
 var ErrCallOnSubscribeDuplicated = errors.New("call OnSubscribe duplicated")
 var EmptySubscription = emptySubscription{}
 
-func TryRecoverError(re interface{}) error {
-	if re == nil {
-		return nil
-	}
-	switch e := re.(type) {
-	case error:
-		return e
-	case string:
-		return errors.New(e)
-	default:
-		return fmt.Errorf("%s", e)
-	}
-}
-
 type emptySubscription struct {
 }
 

internal/misc_test.go

@@ -1,20 +1,12 @@
 package internal_test
 
 import (
-	"errors"
 	"testing"
 
 	"github.com/jjeffcaii/reactor-go/internal"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestTryRecoverError(t *testing.T) {
-	fakeErr := errors.New("fake error")
-	assert.Equal(t, fakeErr, internal.TryRecoverError(fakeErr))
-	assert.Error(t, internal.TryRecoverError("fake error"))
-	assert.Error(t, internal.TryRecoverError(123))
-}
-
 func TestEmptySubscription(t *testing.T) {
 	assert.NotPanics(t, func() {
 		internal.EmptySubscription.Cancel()

internal/subscribers/switch_if_empty.go

@@ -2,6 +2,7 @@ package subscribers
 
 import (
 	"context"
+	"errors"
 
 	"github.com/jjeffcaii/reactor-go"
 )
@@ -54,12 +55,17 @@ func (s *SwitchIfEmptySubscriber) OnSubscribe(ctx context.Context, su reactor.Su
 }
 
 func (s *SwitchIfEmptySubscriber) OnComplete() {
-	if !s.nextOnce {
-		s.nextOnce = true
-		s.other.SubscribeWith(s.ctx, s)
-	} else {
+	if s.nextOnce {
 		s.actual.OnComplete()
+		return
 	}
+	s.nextOnce = true
+	if s.other == nil {
+		s.actual.OnError(errors.New("the alternative SwitchIfEmpty Mono is nil"))
+	} else {
+		s.other.SubscribeWith(s.ctx, s)
+	}
+
 }
 
 func NewSwitchIfEmptySubscriber(alternative reactor.RawPublisher, actual reactor.Subscriber) *SwitchIfEmptySubscriber {

internal/subscribers/switch_value_if_error.go

@@ -41,7 +41,9 @@ func (s *SwitchValueIfErrorSubscriber) Cancel() {
 func (s *SwitchValueIfErrorSubscriber) OnError(err error) {
 	if atomic.AddInt32(&s.errorCalls, 1) == 1 {
 		hooks.Global().OnErrorDrop(err)
-		s.actual.OnNext(s.v)
+		if s.v != nil {
+			s.actual.OnNext(s.v)
+		}
 		s.actual.OnComplete()
 	} else {
 		s.actual.OnError(err)

mono/mono_create.go

@@ -2,17 +2,14 @@ package mono
 
 import (
 	"context"
-	"errors"
 	"sync"
 	"sync/atomic"
 
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/hooks"
-	"github.com/jjeffcaii/reactor-go/internal"
+	"github.com/pkg/errors"
 )
 
-var _errRunSinkFailed = errors.New("execute creation func failed")
-
 var _sinkPool = sync.Pool{
 	New: func() interface{} {
 		return new(sink)
@@ -49,8 +46,14 @@ func newMonoCreate(gen func(context.Context, Sink)) monoCreate {
 	return monoCreate{
 		sinker: func(ctx context.Context, sink Sink) {
 			defer func() {
-				if e := recover(); e != nil {
-					sink.Error(_errRunSinkFailed)
+				rec := recover()
+				if rec == nil {
+					return
+				}
+				if e, ok := rec.(error); ok {
+					sink.Error(errors.WithStack(e))
+				} else {
+					sink.Error(errors.Errorf("%v", rec))
 				}
 			}()
 
@@ -103,8 +106,14 @@ func (s *sink) Error(err error) {
 
 func (s *sink) Next(v Any) {
 	defer func() {
-		if err := internal.TryRecoverError(recover()); err != nil {
-			s.Error(err)
+		rec := recover()
+		if rec == nil {
+			return
+		}
+		if e, ok := rec.(error); ok {
+			s.Error(errors.WithStack(e))
+		} else {
+			s.Error(errors.Errorf("%v", rec))
 		}
 	}()
 	s.actual.OnNext(v)

mono/mono_just.go

@@ -6,7 +6,7 @@ import (
 	"sync/atomic"
 
 	"github.com/jjeffcaii/reactor-go"
-	"github.com/jjeffcaii/reactor-go/internal"
+	"github.com/pkg/errors"
 )
 
 var _justSubscriptionPool = sync.Pool{
@@ -64,10 +64,15 @@ func (j *justSubscription) Request(n int) {
 		return
 	}
 	defer func() {
-		if err := internal.TryRecoverError(recover()); err != nil {
-			j.actual.OnError(err)
-		} else {
+		rec := recover()
+		if rec == nil {
 			j.actual.OnComplete()
+			return
+		}
+		if e, ok := rec.(error); ok {
+			j.actual.OnError(errors.WithStack(e))
+		} else {
+			j.actual.OnError(errors.Errorf("%v", rec))
 		}
 	}()
 	j.actual.OnNext(j.parent.value)

mono/mono_test.go

@@ -2,7 +2,6 @@ package mono_test
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sync/atomic"
 	"testing"
@@ -12,6 +11,7 @@ import (
 	"github.com/jjeffcaii/reactor-go/hooks"
 	"github.com/jjeffcaii/reactor-go/mono"
 	"github.com/jjeffcaii/reactor-go/scheduler"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -144,7 +144,7 @@ func testPanic(m mono.Mono, t *testing.T) {
 		in.DoOnError(func(e error) {
 			catches = e
 		}).Subscribe(context.Background())
-		assert.Equal(t, fakeErr, catches, "not that error")
+		assert.Equal(t, fakeErr, errors.Cause(catches), "not that error")
 	}
 	checker(m.DoOnNext(func(v Any) error {
 		return fakeErr

mono/mono_zip_test.go

@@ -121,3 +121,41 @@ func TestZip_context(t *testing.T) {
 	assert.Error(t, err)
 	assert.True(t, reactor.IsCancelledError(err))
 }
+
+func TestZip_EdgeCase(t *testing.T) {
+	var (
+		nextCnt     = new(int32)
+		completeCnt = new(int32)
+		errorCnt    = new(int32)
+	)
+	mono.Zip(mono.JustOneshot("1"), mono.JustOneshot("2")).
+		FlatMap(func(any reactor.Any) mono.Mono {
+			if any != nil {
+				return mono.Zip(mono.JustOneshot("333"), mono.JustOneshot("44444444")).
+					Filter(func(any reactor.Any) bool {
+						panic("fake panic")
+					}).
+					Map(func(any reactor.Any) (reactor.Any, error) {
+						panic("ddddddd")
+					})
+			}
+			return mono.JustOneshot("dddd")
+		}).
+		Subscribe(context.Background(),
+			reactor.OnNext(func(v reactor.Any) error {
+				atomic.AddInt32(nextCnt, 1)
+				return nil
+			}),
+			reactor.OnError(func(e error) {
+				atomic.AddInt32(errorCnt, 1)
+				t.Logf("%v", e)
+			}),
+			reactor.OnComplete(func() {
+				atomic.AddInt32(completeCnt, 1)
+			}),
+		)
+
+	assert.Equal(t, int32(0), atomic.LoadInt32(nextCnt), "next count should be zero")
+	assert.Equal(t, int32(1), atomic.LoadInt32(errorCnt), "error count should be 1")
+	assert.Equal(t, int32(0), atomic.LoadInt32(completeCnt), "complete count should be zero")
+}

mono/op_filter.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/internal"
+	"github.com/pkg/errors"
 )
 
 type filterSubscriber struct {
@@ -46,8 +47,14 @@ func (f *filterSubscriber) OnError(err error) {
 
 func (f *filterSubscriber) OnNext(v Any) {
 	defer func() {
-		if err := internal.TryRecoverError(recover()); err != nil {
-			f.OnError(err)
+		rec := recover()
+		if rec == nil {
+			return
+		}
+		if e, ok := rec.(error); ok {
+			f.OnError(errors.WithStack(e))
+		} else {
+			f.OnError(errors.Errorf("%v", rec))
 		}
 	}()
 	if f.predicate(v) {

mono/op_flatmap.go

@@ -5,6 +5,7 @@ import (
 	"sync/atomic"
 
 	"github.com/jjeffcaii/reactor-go"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -71,11 +72,34 @@ func (p *flatMapSubscriber) OnNext(v Any) {
 	if atomic.LoadInt32(&p.stat) != 0 {
 		return
 	}
-	m := p.mapper(v)
+	nextMono, err := p.computeNextMono(v)
+	if err != nil {
+		p.actual.OnError(err)
+		return
+	}
 	inner := &innerFlatMapSubscriber{
 		parent: p,
 	}
-	m.SubscribeWith(p.ctx, inner)
+	nextMono.SubscribeWith(p.ctx, inner)
+}
+
+func (p *flatMapSubscriber) computeNextMono(v Any) (next Mono, err error) {
+	defer func() {
+		rec := recover()
+		if rec == nil {
+			return
+		}
+		if e, ok := rec.(error); ok {
+			err = errors.WithStack(e)
+		} else {
+			err = errors.Errorf("%v", rec)
+		}
+	}()
+	next = p.mapper(v)
+	if next == nil {
+		err = errors.New("the FlatMap result is nil")
+	}
+	return
 }
 
 func (p *flatMapSubscriber) OnSubscribe(ctx context.Context, s reactor.Subscription) {

mono/op_map.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/jjeffcaii/reactor-go"
+	"github.com/pkg/errors"
 )
 
 var _mapSubscriberPool = sync.Pool{
@@ -68,12 +69,25 @@ func (m *mapSubscriber) OnError(err error) {
 
 func (m *mapSubscriber) OnNext(v Any) {
 	if m == nil || m.actual == nil || m.t == nil {
+		// TODO:
 		return
 	}
-	if o, err := m.t(v); err != nil {
+	defer func() {
+		rec := recover()
+		if rec == nil {
+			return
+		}
+		if e, ok := rec.(error); ok {
+			m.actual.OnError(errors.WithStack(e))
+		} else {
+			m.actual.OnError(errors.Errorf("%v", rec))
+		}
+	}()
+
+	if transformed, err := m.t(v); err != nil {
 		m.actual.OnError(err)
 	} else {
-		m.actual.OnNext(o)
+		m.actual.OnNext(transformed)
 	}
 }