Support Go1.18 generics for all interface{} operations

Author: cschleiden

.gitignore

@@ -1,2 +1,3 @@
 *.sqlite
-*.sqlite-journal
+*.sqlite-journal
+vendor

.vscode/settings.json

@@ -33,5 +33,9 @@
   },
   "github-actions.workflows.pinned.workflows": [
     ".github/workflows/go.yml"
-  ]
+  ],
+  "go.alternateTools": {
+    "go": "/Users/cschleiden/sdk/go1.18rc1/bin/go",
+  },
+  "go.testExplorer.showDynamicSubtestsInEditor": true,
 }

go.mod

@@ -1,6 +1,6 @@
 module github.com/cschleiden/go-workflows
 
-go 1.17
+go 1.18
 
 require (
 	github.com/go-sql-driver/mysql v1.6.0

internal/sync/channel.go

@@ -2,74 +2,76 @@ package sync
 
 import (
 	"github.com/cschleiden/go-workflows/internal/converter"
-	"github.com/pkg/errors"
 )
 
-type Channel interface {
-	Send(ctx Context, v interface{})
+type Channel[T any] interface {
+	Send(ctx Context, v T)
 
-	SendNonblocking(ctx Context, v interface{}) (ok bool)
+	SendNonblocking(ctx Context, v T) (ok bool)
 
-	Receive(ctx Context, vptr interface{}) (more bool)
+	Receive(ctx Context) (v T, ok bool)
 
-	ReceiveNonblocking(ctx Context, vptr interface{}) (more bool)
+	ReceiveNonblocking(ctx Context) (v T, ok bool)
 
 	Close()
 }
 
-type ChannelInternal interface {
+type ChannelInternal[T any] interface {
 	Closed() bool
 
-	ReceiveNonBlocking(ctx Context, cb func(v interface{})) (ok bool)
+	ReceiveNonBlocking(ctx Context) (v T, ok bool)
 
-	AddReceiveCallback(cb func(v interface{}))
+	AddReceiveCallback(cb func(v T, ok bool))
 }
 
-func NewChannel() Channel {
-	return &channel{
-		c:         make([]interface{}, 0),
+// Ensure channel implementation support internal interface
+var _ ChannelInternal[struct{}] = (*channel[struct{}])(nil)
+
+func NewChannel[T any]() Channel[T] {
+	return &channel[T]{
+		c:         make([]T, 0),
 		converter: converter.DefaultConverter,
 	}
 }
 
-func NewBufferedChannel(size int) Channel {
-	return &channel{
-		c:         make([]interface{}, 0, size),
+func NewBufferedChannel[T any](size int) Channel[T] {
+	return &channel[T]{
+		c:         make([]T, 0, size),
 		size:      size,
 		converter: converter.DefaultConverter,
 	}
 }
 
-type channel struct {
-	c         []interface{}
-	receivers []func(interface{})
-	senders   []func() interface{}
+type channel[T any] struct {
+	c         []T
+	receivers []func(value T, ok bool)
+	senders   []func() T
 	closed    bool
 	size      int
 	converter converter.Converter
 }
 
-var _ Channel = (*channel)(nil)
-var _ ChannelInternal = (*channel)(nil)
-
-func (c *channel) Close() {
+func (c *channel[T]) Close() {
 	c.closed = true
 
 	// If there are still blocked senders, error
 	if len(c.senders) > 0 {
 		panic("send on closed channel")
 	}
 
+	// TODO: Drain buffered values
 	for len(c.receivers) > 0 {
 		r := c.receivers[0]
 		c.receivers[0] = nil
 		c.receivers = c.receivers[1:]
 
-		r(nil)
+		// Send zero value to pending receiver
+		var v T
+		r(v, false)
 	}
 }
 
-func (c *channel) Send(ctx Context, v interface{}) {
+func (c *channel[T]) Send(ctx Context, v T) {
 	cr := getCoState(ctx)
 
 	addedSender := false
@@ -84,50 +86,48 @@ func (c *channel) Send(ctx Context, v interface{}) {
 		if !addedSender {
 			addedSender = true
 
-			cb := func() interface{} {
+			cb := func() T {
 				sentValue = true
 				return v
 			}
 
 			c.senders = append(c.senders, cb)
 		}
 
+		// No waiting receiver, yield
 		cr.Yield()
 
+		// Was our sender called while we yielded? If so, we can return
 		if sentValue {
 			cr.MadeProgress()
 			return
 		}
 	}
 }
 
-func (c *channel) SendNonblocking(ctx Context, v interface{}) bool {
+func (c *channel[T]) SendNonblocking(ctx Context, v T) bool {
 	return c.trySend(v)
 }
 
-func (c *channel) Receive(ctx Context, vptr interface{}) (more bool) {
+func (c *channel[T]) Receive(ctx Context) (v T, ok bool) {
 	cr := getCoState(ctx)
 
 	addedListener := false
 	receivedValue := false
 
 	for {
 		// Try to receive from buffered channel or blocked sender
-		if c.tryReceive(vptr) {
+		if v, ok, rok := c.tryReceive(); rok {
 			cr.MadeProgress()
-			return !c.closed
+			return v, ok
 		}
 
 		// Register handler to receive value once
 		if !addedListener {
-			cb := func(v interface{}) {
+			cb := func(rv T, rok bool) {
 				receivedValue = true
-
-				if vptr != nil {
-					if err := converter.AssignValue(c.converter, v, vptr); err != nil {
-						panic(err)
-					}
-				}
+				v = rv
+				ok = rok
 			}
 
 			c.receivers = append(c.receivers, cb)
@@ -139,25 +139,30 @@ func (c *channel) Receive(ctx Context, vptr interface{}) (more bool) {
 		// If we received a value via the callback, return
 		if receivedValue {
 			cr.MadeProgress()
-			return !c.closed
+			return v, ok
 		}
 	}
 }
 
-func (c *channel) ReceiveNonblocking(ctx Context, vptr interface{}) (ok bool) {
-	return c.tryReceive(vptr)
+func (c *channel[T]) ReceiveNonblocking(ctx Context) (T, bool) {
+	if v, ok, rok := c.tryReceive(); rok {
+		return v, ok
+	}
+
+	var z T
+	return z, false
 }
 
-func (c *channel) hasValue() bool {
+func (c *channel[T]) hasValue() bool {
 	return len(c.c) > 0
 }
 
-func (c *channel) canReceive() bool {
+func (c *channel[T]) canReceive() bool {
 	return c.hasValue() || len(c.senders) > 0 || c.closed
 }
 
-func (c *channel) trySend(v interface{}) bool {
-	// If closed, we can't send, exit.
+func (c *channel[T]) trySend(v T) bool {
+	// If closed, we can't send, panic.
 	if c.closed {
 		panic("channel closed")
 	}
@@ -168,7 +173,9 @@ func (c *channel) trySend(v interface{}) bool {
 		r := c.receivers[0]
 		c.receivers[0] = nil
 		c.receivers = c.receivers[1:]
-		r(v)
+
+		r(v, true)
+
 		return true
 	}
 
@@ -182,72 +189,52 @@ func (c *channel) trySend(v interface{}) bool {
 	return false
 }
 
-func (c *channel) tryReceive(vptr interface{}) bool {
+func (c *channel[T]) tryReceive() (v T, ok bool, rok bool) {
 	// If channel is buffered, return value if available
 	if c.hasValue() {
-		v := c.c[0]
+		v = c.c[0]
 		c.c = c.c[1:]
 
-		if vptr != nil {
-			if err := converter.AssignValue(c.converter, v, vptr); err != nil {
-				panic(errors.Wrap(err, "could not assign value when receiving from channel"))
-			}
-		}
-
-		return true
+		return v, true, true
 	}
 
 	// If channel has been closed and no values in buffer (if buffered) return zero
 	// element
 	if c.closed {
-		if vptr != nil {
-			if err := converter.AssignValue(c.converter, nil, vptr); err != nil {
-				panic(err)
-			}
-		}
-
-		return true
+		var z T
+		return z, false, true
 	}
 
+	// Any blocked senders? If so, receive from the first one
 	if len(c.senders) > 0 {
 		s := c.senders[0]
 		c.senders[0] = nil
 		c.senders = c.senders[1:]
 
-		v := s()
-
-		if vptr != nil {
-			if err := converter.AssignValue(c.converter, v, vptr); err != nil {
-				panic(err)
-			}
-		}
-
-		return true
+		return s(), true, true
 	}
 
-	return false
+	// Could not receive value
+	return v, ok, false
 }
 
-func (c *channel) hasCapacity() bool {
+func (c *channel[T]) hasCapacity() bool {
 	return len(c.c) < c.size
 }
 
-func (c *channel) AddReceiveCallback(cb func(v interface{})) {
+func (c *channel[T]) AddReceiveCallback(cb func(v T, ok bool)) {
 	c.receivers = append(c.receivers, cb)
 }
 
-func (c *channel) ReceiveNonBlocking(ctx Context, cb func(v interface{})) (ok bool) {
-	var vptr interface{}
-	if c.tryReceive(vptr) {
-		cb(vptr)
-		return true
+func (c *channel[T]) ReceiveNonBlocking(ctx Context) (T, bool) {
+	if v, ok, rok := c.tryReceive(); rok {
+		return v, ok
 	}
 
-	c.AddReceiveCallback(cb)
-
-	return false
+	var z T
+	return z, false
 }
 
-func (c *channel) Closed() bool {
+func (c *channel[T]) Closed() bool {
 	return c.closed
 }