chore: ut (#32)

* chore: ut

* fix: more ut

internal/misc.go

@@ -5,7 +5,7 @@ import (
 )
 
 var ErrCallOnSubscribeDuplicated = errors.New("call OnSubscribe duplicated")
-var EmptySubscription = emptySubscription{}
+var EmptySubscription emptySubscription
 
 type emptySubscription struct {
 }

internal/misc_test.go

@@ -13,3 +13,9 @@ func TestEmptySubscription(t *testing.T) {
 		internal.EmptySubscription.Request(1)
 	})
 }
+
+func TestSafeCloseDone(t *testing.T) {
+	c := make(chan struct{})
+	assert.True(t, internal.SafeCloseDone(c))
+	assert.False(t, internal.SafeCloseDone(c))
+}

internal/subscribers/do_finally.go

@@ -8,17 +8,31 @@ import (
 	"github.com/jjeffcaii/reactor-go"
 )
 
-var _doFinallySubscriberPool = sync.Pool{
-	New: func() interface{} {
-		return new(DoFinallySubscriber)
-	},
-}
-
 var (
 	_ reactor.Subscriber = (*DoFinallySubscriber)(nil)
 	_ reactor.Disposable = (*DoFinallySubscriber)(nil)
 )
 
+var globalDoFinallySubscriberPool doFinallySubscriberPool
+
+type doFinallySubscriberPool struct {
+	inner sync.Pool
+}
+
+func (p *doFinallySubscriberPool) get() *DoFinallySubscriber {
+	if exist, _ := p.inner.Get().(*DoFinallySubscriber); exist != nil {
+		return exist
+	}
+	return &DoFinallySubscriber{}
+}
+
+func (p *doFinallySubscriberPool) put(s *DoFinallySubscriber) {
+	if s == nil {
+		return
+	}
+	p.inner.Put(s)
+}
+
 type DoFinallySubscriber struct {
 	actual    reactor.Subscriber
 	onFinally reactor.FnOnFinally
@@ -30,11 +44,11 @@ func (d *DoFinallySubscriber) Dispose() {
 	d.actual = nil
 	d.onFinally = nil
 	d.s = nil
-	_doFinallySubscriberPool.Put(d)
+	globalDoFinallySubscriberPool.put(d)
 }
 
 func NewDoFinallySubscriber(actual reactor.Subscriber, onFinally reactor.FnOnFinally) *DoFinallySubscriber {
-	s := _doFinallySubscriberPool.Get().(*DoFinallySubscriber)
+	s := globalDoFinallySubscriberPool.get()
 	s.actual = actual
 	s.onFinally = onFinally
 	atomic.StoreInt32(&s.done, 0)

internal/subscribers/do_finally_test.go

@@ -0,0 +1,114 @@
+package subscribers
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/jjeffcaii/reactor-go"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDoFinallySubscriber(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	su := NewMockSubscription(ctrl)
+	su.EXPECT().Request(gomock.Any()).Times(1)
+	su.EXPECT().Cancel().Times(0)
+
+	var prevOperation string
+
+	actual := NewMockSubscriber(ctrl)
+	actual.EXPECT().
+		OnSubscribe(gomock.Any(), gomock.Any()).
+		Do(func(ctx context.Context, su reactor.Subscription) {
+			su.Request(reactor.RequestInfinite)
+			prevOperation = "OnSubscribe"
+		}).
+		Times(1)
+	actual.EXPECT().OnNext(gomock.Any()).Do(func(v reactor.Any) {
+		prevOperation = "OnNext"
+	}).Times(1)
+	actual.EXPECT().OnError(gomock.Any()).Times(0)
+	actual.EXPECT().OnComplete().Do(func() {
+		prevOperation = "OnComplete"
+	}).Times(1)
+
+	s := NewDoFinallySubscriber(actual, func(signal reactor.SignalType) {
+		prevOperation = "DoFinally"
+		assert.Equal(t, reactor.SignalTypeComplete, signal)
+	})
+	s.OnSubscribe(context.Background(), su)
+	s.OnNext(42)
+	s.OnComplete()
+
+	assert.Equal(t, "DoFinally", prevOperation)
+
+	prevOperation = ""
+	su = NewMockSubscription(ctrl)
+	su.EXPECT().Request(gomock.Any()).Times(1)
+	su.EXPECT().Cancel().Times(0)
+	actual = NewMockSubscriber(ctrl)
+	actual.EXPECT().
+		OnSubscribe(gomock.Any(), gomock.Any()).
+		Do(func(ctx context.Context, su reactor.Subscription) {
+			su.Request(reactor.RequestInfinite)
+			prevOperation = "OnSubscribe"
+		}).
+		Times(1)
+	actual.EXPECT().OnNext(gomock.Any()).Do(func(v reactor.Any) {
+		prevOperation = "OnNext"
+	}).Times(0)
+	actual.EXPECT().OnError(gomock.Any()).Do(func(e error) {
+		prevOperation = "OnError"
+	}).Times(1)
+	actual.EXPECT().OnComplete().Do(func() {
+		prevOperation = "OnComplete"
+	}).Times(0)
+	s = NewDoFinallySubscriber(actual, func(signal reactor.SignalType) {
+		prevOperation = "DoFinally"
+		assert.Equal(t, reactor.SignalTypeError, signal)
+	})
+
+	fakeErr := errors.New("fake error")
+	s.OnSubscribe(context.Background(), su)
+	s.OnError(fakeErr)
+
+	assert.Equal(t, "DoFinally", prevOperation)
+}
+
+func TestDoFinallySubscriberPool_Put(t *testing.T) {
+	assert.NotPanics(t, func() {
+		globalDoFinallySubscriberPool.put(nil)
+	})
+}
+
+func TestDoFinallySubscriber_Cancel(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	su := NewMockSubscription(ctrl)
+	su.EXPECT().Request(gomock.Any()).Times(0)
+	su.EXPECT().Cancel().Times(1)
+
+	actual := NewMockSubscriber(ctrl)
+	actual.EXPECT().OnSubscribe(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, su reactor.Subscription) {
+		su.Cancel()
+	}).Times(1)
+	actual.EXPECT().OnError(gomock.Any()).AnyTimes()
+	actual.EXPECT().OnNext(gomock.Any()).Times(0)
+	actual.EXPECT().OnComplete().Times(0)
+
+	var doFinallyCalls int
+
+	s := NewDoFinallySubscriber(actual, func(signal reactor.SignalType) {
+		assert.Equal(t, reactor.SignalTypeCancel, signal)
+		doFinallyCalls++
+	})
+	s.Cancel()
+	s.OnSubscribe(context.Background(), su)
+
+	assert.Equal(t, 1, doFinallyCalls)
+}