Merge pull request #124 from cschleiden/add-send-select-case

Add Send case to the Select function

Author: cschleiden

README.md

@@ -410,6 +410,8 @@ Due its non-deterministic behavior you must not use a `select` statement in work
 var f1 workflow.Future[int]
 var c workflow.Channel[int]
 
+value := 42
+
 workflow.Select(
 	ctx,
 	workflow.Await(f1, func (ctx workflow.Context, f Future[int]) {
@@ -419,6 +421,9 @@ workflow.Select(
 	workflow.Receive(c, func (ctx workflow.Context, v int, ok bool) {
 		// use v
 	}),
+	workflow.Send(c, &value, func (ctx workflow.Context) {
+		// value has been sent to the channel
+	}),
 	workflow.Default(ctx, func (ctx workflow.Context) {
 		// ...
 	})

internal/sync/channel.go

@@ -155,6 +155,14 @@ func (c *channel[T]) canReceive() bool {
 	return c.hasValue() || len(c.senders) > 0 || c.closed
 }
 
+func (c *channel[T]) canSend() bool {
+	if c.closed {
+		return false
+	}
+
+	return len(c.receivers) > 0 || c.hasCapacity()
+}
+
 func (c *channel[T]) trySend(v T) bool {
 	// If closed, we can't send, panic.
 	if c.closed {

internal/sync/selector.go

@@ -12,8 +12,16 @@ func Await[T any](f Future[T], handler func(ctx Context, f Future[T])) SelectCas
 	}
 }
 
+func Send[T any](c Channel[T], v *T, handler func(ctx Context)) SelectCase {
+	return &channelSendCase[T]{
+		c:  c.(*channel[T]),
+		v:  v,
+		fn: handler,
+	}
+}
+
 func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase {
-	return &channelCase[T]{
+	return &channelReceiveCase[T]{
 		c:  c.(*channel[T]),
 		fn: handler,
 	}
@@ -55,18 +63,33 @@ func (fc *futureCase[T]) Handle(ctx Context) {
 	fc.fn(ctx, fc.f)
 }
 
-type channelCase[T any] struct {
+type channelReceiveCase[T any] struct {
 	c  *channel[T]
 	fn func(Context, T, bool)
 }
 
-func (cc *channelCase[T]) Ready() bool {
-	return cc.c.canReceive()
+func (crc *channelReceiveCase[T]) Ready() bool {
+	return crc.c.canReceive()
+}
+
+func (crc *channelReceiveCase[T]) Handle(ctx Context) {
+	v, ok := crc.c.Receive(ctx)
+	crc.fn(ctx, v, ok)
+}
+
+type channelSendCase[T any] struct {
+	c  *channel[T]
+	v  *T
+	fn func(Context)
+}
+
+func (csc *channelSendCase[T]) Ready() bool {
+	return csc.c.canSend()
 }
 
-func (cc *channelCase[T]) Handle(ctx Context) {
-	v, ok := cc.c.Receive(ctx)
-	cc.fn(ctx, v, ok)
+func (csc *channelSendCase[T]) Handle(ctx Context) {
+	csc.c.Send(ctx, *csc.v)
+	csc.fn(ctx)
 }
 
 type defaultCase struct {

internal/sync/selector_test.go

@@ -111,7 +111,7 @@ func Test_FutureSelector_DefaultCase(t *testing.T) {
 	require.True(t, defaultHandled)
 }
 
-func Test_ChannelSelector_Select(t *testing.T) {
+func Test_ChannelSelector_Receive(t *testing.T) {
 	c := NewChannel[int]()
 
 	reachedEnd := false
@@ -148,3 +148,82 @@ func Test_ChannelSelector_Select(t *testing.T) {
 	require.True(t, reachedEnd)
 	require.Equal(t, 42, r)
 }
+
+func Test_ChannelSelector_SendBlocking(t *testing.T) {
+	c := NewChannel[int]()
+	ctx := Background()
+
+	input := 42
+
+	cs := NewCoroutine(ctx, func(ctx Context) error {
+		Select(
+			ctx,
+			Send(c, &input, func(ctx Context) {
+				// Element was sent
+			}),
+		)
+
+		return nil
+	})
+
+	cs.Execute()
+
+	var v int
+	var ok bool
+
+	cr := NewCoroutine(ctx, func(ctx Context) error {
+		v, ok = c.Receive(ctx)
+
+		return nil
+	})
+	// Register receiver
+	cr.Execute()
+
+	// Try to Select again, this time it sends
+	cs.Execute()
+
+	// Allow receiver to finish
+	cr.Execute()
+
+	require.True(t, cs.Finished())
+	require.True(t, cr.Finished())
+	require.Equal(t, 42, v)
+	require.True(t, ok)
+}
+
+func Test_ChannelSelector_SendNonBlocking(t *testing.T) {
+	c := NewBufferedChannel[int](1)
+	ctx := Background()
+
+	cs := NewCoroutine(ctx, func(ctx Context) error {
+		input := 42
+		Select(
+			ctx,
+			Send(c, &input, func(ctx Context) {
+				// Element was sent
+			}),
+		)
+
+		return nil
+	})
+
+	cs.Execute()
+	require.True(t, cs.Finished())
+
+	cs = NewCoroutine(ctx, func(ctx Context) error {
+		input := 23
+		Select(
+			ctx,
+			Send(c, &input, func(ctx Context) {
+				// Element was sent
+			}),
+		)
+
+		return nil
+	})
+
+	cs.Execute()
+
+	// Channel does not have capacity, Select blocks
+	require.False(t, cs.Finished())
+}

workflow/sync.go

@@ -5,36 +5,46 @@ import (
 )
 
 type Context = sync.Context
-type WaitGroup = sync.WaitGroup
 
 var Canceled = sync.Canceled
 
+type WaitGroup = sync.WaitGroup
+
+func NewWaitGroup() WaitGroup {
+	return sync.NewWaitGroup()
+}
+
+// Go spawns a workflow goroutine
 func Go(ctx Context, f func(ctx Context)) {
 	sync.Go(ctx, f)
 }
 
 type SelectCase = sync.SelectCase
 
+// Select is the workflow-save equivalent of the select statement.
 func Select(ctx Context, cases ...SelectCase) {
 	sync.Select(ctx, cases...)
 }
 
+// Await calls the provided handler when the given future is ready.
 func Await[T any](f Future[T], handler func(Context, Future[T])) SelectCase {
 	return sync.Await[T](f, func(ctx sync.Context, f sync.Future[T]) {
 		handler(ctx, f)
 	})
 }
 
+// Receive calls the provided handler if the given channel can receive a value. The handler receives
+// the received value, and the ok flag indicating whether the value was received or the channel was closed.
 func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase {
-	return sync.Receive[T](c, func(ctx sync.Context, v T, ok bool) {
-		handler(ctx, v, ok)
-	})
+	return sync.Receive[T](c, handler)
 }
 
-func Default(handler func(Context)) SelectCase {
-	return sync.Default(handler)
+// Send calls the provided handler if the given value can be sent to the channel.
+func Send[T any](c Channel[T], value *T, handler func(ctx Context)) SelectCase {
+	return sync.Send[T](c, value, handler)
 }
 
-func NewWaitGroup() WaitGroup {
-	return sync.NewWaitGroup()
+// Default calls the given handler if none of the other cases match.
+func Default(handler func(Context)) SelectCase {
+	return sync.Default(handler)
 }