fix: mono block race problem (#26)

Author: jjeffcaii

flux/flux_create.go

@@ -43,7 +43,7 @@ func newFluxCreate(c func(ctx context.Context, sink Sink), options ...CreateOpti
 func (fc fluxCreate) SubscribeWith(ctx context.Context, s reactor.Subscriber) {
 	select {
 	case <-ctx.Done():
-		s.OnError(ctx.Err())
+		s.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		var sink interface {
 			reactor.Subscription

flux/flux_interval.go

@@ -57,9 +57,7 @@ func (is *intervalSubscription) run(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			if err := ctx.Err(); err != nil {
-				is.actual.OnError(err)
-			}
+			is.actual.OnError(reactor.NewContextError(ctx.Err()))
 			return
 		case <-is.done:
 			return

internal/buffer/buffer.go internal/buffer/unbounded.go

@@ -18,7 +18,9 @@
 // Package buffer provides an implementation of an unbounded buffer.
 package buffer
 
-import "sync"
+import (
+	"sync"
+)
 
 // Unbounded is an implementation of an unbounded buffer which does not use
 // extra goroutines. This is typically used for passing updates from one entity
@@ -34,9 +36,10 @@ import "sync"
 // defining a new type specific implementation of this buffer is preferred. See
 // internal/transport/transport.go for an example of this.
 type Unbounded struct {
-	c       chan interface{}
-	mu      sync.Mutex
-	backlog []interface{}
+	c        chan interface{}
+	mu       sync.Mutex
+	backlog  []interface{}
+	disposed bool
 }
 
 // NewUnbounded returns a new instance of Unbounded.
@@ -47,35 +50,46 @@ func NewUnbounded() *Unbounded {
 // Put adds t to the unbounded buffer.
 func (b *Unbounded) Put(t interface{}) (ok bool) {
 	b.mu.Lock()
-	defer func() {
-		ok = recover() == nil
+
+	if b.disposed {
 		b.mu.Unlock()
-	}()
+		return
+	}
+
+	ok = true
+
 	if len(b.backlog) == 0 {
 		select {
 		case b.c <- t:
+			b.mu.Unlock()
 			return
 		default:
 		}
 	}
 	b.backlog = append(b.backlog, t)
+	b.mu.Unlock()
 	return
 }
 
 // Load sends the earliest buffered data, if any, onto the read channel
 // returned by Get(). Users are expected to call this every time they read a
 // value from the read channel.
-func (b *Unbounded) Load() {
+func (b *Unbounded) Load() (n int) {
 	b.mu.Lock()
 	if len(b.backlog) > 0 {
 		select {
 		case b.c <- b.backlog[0]:
 			b.backlog[0] = nil
 			b.backlog = b.backlog[1:]
+			n = 1
 		default:
 		}
+	} else if b.disposed {
+		b.close()
+		n = -1
 	}
 	b.mu.Unlock()
+	return
 }
 
 // Get returns a read channel on which values added to the buffer, via Put(),
@@ -87,6 +101,20 @@ func (b *Unbounded) Get() <-chan interface{} {
 	return b.c
 }
 
+// Dispose mark current Unbounded as disposed.
 func (b *Unbounded) Dispose() {
+	b.mu.Lock()
+	b.disposed = true
+	if len(b.backlog) == 0 {
+		b.close()
+	}
+	b.mu.Unlock()
+}
+
+func (b *Unbounded) close() (ok bool) {
+	defer func() {
+		ok = recover() == nil
+	}()
 	close(b.c)
+	return
 }

internal/buffer/unbounded_test.go

@@ -0,0 +1,61 @@
+package buffer_test
+
+import (
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/jjeffcaii/reactor-go/internal/buffer"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewUnbounded(t *testing.T) {
+	u := buffer.NewUnbounded()
+
+	go func() {
+		assert.True(t, u.Put("foo"))
+		assert.True(t, u.Put("bar"))
+		assert.True(t, u.Put("qux"))
+
+		u.Dispose()
+		assert.False(t, u.Put("must failed"))
+	}()
+
+	done := make(chan struct{})
+	var read []string
+
+	go func() {
+		defer close(done)
+		for next := range u.Get() {
+			read = append(read, next.(string))
+			u.Load()
+		}
+	}()
+
+	<-done
+
+	assert.Equal(t, "foo,bar,qux", strings.Join(read, ","), "result doesn't match")
+}
+
+func TestEmptyUnbounded(t *testing.T) {
+	u := buffer.NewUnbounded()
+
+	done := make(chan struct{})
+	cnt := new(int32)
+
+	go func() {
+		defer close(done)
+		for range u.Get() {
+			atomic.AddInt32(cnt, 1)
+			u.Load()
+		}
+	}()
+
+	go func() {
+		u.Dispose()
+	}()
+
+	<-done
+
+	assert.Zero(t, atomic.LoadInt32(cnt))
+}

internal/stopwatch/Stopwatch.go

@@ -54,7 +54,7 @@ func (b *blockFirstSubscriber) OnNext(any reactor.Any) {
 func (b *blockFirstSubscriber) OnSubscribe(ctx context.Context, subscription reactor.Subscription) {
 	select {
 	case <-ctx.Done():
-		b.OnError(reactor.ErrSubscribeCancelled)
+		b.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		b.su = subscription
 		b.su.Request(1)

internal/subscribers/block_first_subscriber.go

@@ -70,7 +70,7 @@ func (b *BlockLastSubscriber) OnNext(value reactor.Any) {
 func (b *BlockLastSubscriber) OnSubscribe(ctx context.Context, subscription reactor.Subscription) {
 	select {
 	case <-ctx.Done():
-		b.OnError(reactor.ErrSubscribeCancelled)
+		b.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		subscription.Request(reactor.RequestInfinite)
 	}

internal/subscribers/block_last_subscriber.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/jjeffcaii/reactor-go"
+	"github.com/jjeffcaii/reactor-go/hooks"
 )
 
 type blockSubscriber struct {
@@ -29,20 +30,26 @@ func (b blockSubscriber) OnComplete() {
 func (b blockSubscriber) OnError(err error) {
 	select {
 	case <-b.done:
+		hooks.Global().OnErrorDrop(err)
 	default:
-		close(b.done)
-		b.c <- reactor.Item{
-			E: err,
+		select {
+		case b.c <- reactor.Item{E: err}:
+		default:
+			hooks.Global().OnErrorDrop(err)
 		}
+		close(b.done)
 	}
 }
 
 func (b blockSubscriber) OnNext(any reactor.Any) {
 	select {
 	case <-b.done:
+		hooks.Global().OnNextDrop(any)
 	default:
-		b.c <- reactor.Item{
-			V: any,
+		select {
+		case b.c <- reactor.Item{V: any}:
+		default:
+			hooks.Global().OnNextDrop(any)
 		}
 	}
 }
@@ -53,7 +60,7 @@ func (b blockSubscriber) OnSubscribe(ctx context.Context, subscription reactor.S
 		go func() {
 			select {
 			case <-ctx.Done():
-				b.OnError(reactor.ErrSubscribeCancelled)
+				b.OnError(reactor.NewContextError(ctx.Err()))
 			case <-b.done:
 			}
 		}()

internal/subscribers/block_subscriber.go

@@ -68,7 +68,7 @@ func (d *DoFinallySubscriber) OnNext(v reactor.Any) {
 func (d *DoFinallySubscriber) OnSubscribe(ctx context.Context, s reactor.Subscription) {
 	select {
 	case <-ctx.Done():
-		d.OnError(reactor.ErrSubscribeCancelled)
+		d.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		d.s = s
 		d.actual.OnSubscribe(ctx, d)

internal/subscribers/do_finally.go

@@ -1,8 +1,8 @@
-default:
-	echo 'Hello, world!'
+alias t := test
+
 test:
-    go test -cover -race -count=1 ./...
+    go test -race -count=1 ./...
 lint:
     golangci-lint run ./...
 fmt:
-    go fmt ./...
+    go fmt ./...

justfile

@@ -8,7 +8,12 @@ import (
 	"github.com/jjeffcaii/reactor-go/scheduler"
 )
 
-type Any = reactor.Any
+// Alias
+type (
+	Any        = reactor.Any
+	Disposable = reactor.Disposable
+)
+
 type FlatMapper = func(reactor.Any) Mono
 
 type Mono interface {

mono/mono.go

@@ -56,7 +56,7 @@ func newMonoCreate(gen func(context.Context, Sink)) monoCreate {
 
 			select {
 			case <-ctx.Done():
-				sink.Error(reactor.ErrSubscribeCancelled)
+				sink.Error(reactor.NewContextError(ctx.Err()))
 			default:
 				gen(ctx, sink)
 			}

mono/mono_create.go

@@ -83,7 +83,7 @@ func (j *justSubscription) Cancel() {
 func (m *monoJust) SubscribeWith(ctx context.Context, sub reactor.Subscriber) {
 	select {
 	case <-ctx.Done():
-		sub.OnError(reactor.ErrSubscribeCancelled)
+		sub.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		su := borrowJustSubscription()
 		su.parent = m

mono/mono_just.go

@@ -143,7 +143,7 @@ func (z *zipInner) OnNext(any reactor.Any) {
 func (z *zipInner) OnSubscribe(ctx context.Context, su reactor.Subscription) {
 	select {
 	case <-ctx.Done():
-		z.OnError(reactor.ErrSubscribeCancelled)
+		z.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		var exist bool
 		z.Lock()
@@ -169,7 +169,7 @@ type monoZip struct {
 func (m *monoZip) SubscribeWith(ctx context.Context, sub reactor.Subscriber) {
 	select {
 	case <-ctx.Done():
-		sub.OnError(reactor.ErrSubscribeCancelled)
+		sub.OnError(reactor.NewContextError(ctx.Err()))
 	default:
 		c := newZipCoordinator(sub, len(m.sources))
 		sub.OnSubscribe(ctx, c)