Added await-signals sample (#236)

README.md

@@ -179,6 +179,9 @@ These samples demonstrate some common control flow patterns using Temporal's Go
 - [**Saga pattern**](https://github.com/temporalio/samples-go/tree/main/saga): This sample demonstrates how to implement
   a saga pattern using golang defer feature.
 
+- [**Await for signal processing**](https://github.com/temporalio/samples-go/tree/main/await-signals): Demonstrates how
+  to process out of order signals processing using `Await` and `AwaitWithTimeout`.
+
 ### Scenario based examples
 
 - [**DSL Workflow**](https://github.com/temporalio/samples-go/tree/master/dsl): Demonstrates how to implement a

await-signals/README.md

@@ -0,0 +1,56 @@
+* The sample demonstrates how to deal with multiple signals that can come out of order and require actions
+* if a certain signal not received in a specified time interval.
+
+This specific sample receives three signals: Signal1, Signal2, Signal3. They have to be processed in the
+sequential order, but they can be received out of order.
+There are two timeouts to enforce.
+The first one is the maximum time between signals.
+The second limits the total time since the first signal received.
+
+A naive implementation of such use case would use a single loop that contains a Selector to listen on three
+signals and a timer. Something like:
+
+	for {
+		selector := workflow.NewSelector(ctx)
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal1"), func(c workflow.ReceiveChannel, more bool) {
+			// Process signal1
+		})
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal2"), func(c workflow.ReceiveChannel, more bool) {
+			// Process signal2
+		}
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal3"), func(c workflow.ReceiveChannel, more bool) {
+			// Process signal3
+		}
+		cCtx, cancel := workflow.WithCancel(ctx)
+		timer := workflow.NewTimer(cCtx, timeToNextSignal)
+		selector.AddFuture(timer, func(f workflow.Future) {
+			// Process timeout
+		})
+ 		selector.Select(ctx)
+	    cancel()
+      // break out of the loop on certain condition
+	}
+
+The above implementation works. But it quickly becomes pretty convoluted if the number of signals
+and rules around order of their arrivals and timeouts increases.
+
+The following example demonstrates an alternative approach. It receives signals in a separate goroutine.
+Each signal handler just updates a correspondent shared variable with the signal data.
+The main workflow function awaits the next step using `workflow.AwaitWithTimeout` using condition composed of
+the shared variables. This makes the main workflow method free from signal callbacks and makes the business logic
+clear.
+
+### Steps to run this sample:
+
+1) You need a Temporal service running. See details in README.md
+2) Run the following command to start the worker
+
+```
+go run await-signals/worker/main.go
+```
+
+3) Run the following command to start the workflow and send signals in random order
+
+```
+go run await-signals/starter/main.go
+```

await-signals/await_signals_workflow.go

@@ -0,0 +1,165 @@
+package await_signals
+
+import (
+	"go.temporal.io/sdk/temporal"
+	"time"
+
+	"go.temporal.io/sdk/workflow"
+)
+
+/**
+ * The sample demonstrates how to deal with multiple signals that can come out of order and require actions
+ * if a certain signal not received in a specified time interval.
+ *
+ * This specific sample receives three signals: Signal1, Signal2, Signal3. They have to be processed in the
+ * sequential order, but they can be received out of order.
+ * There are two timeouts to enforce.
+ * The first one is the maximum time between signals.
+ * The second limits the total time since the first signal received.
+ *
+ * A naive implementation of such use case would use a single loop that contains a Selector to listen on three
+ * signals and a timer. Something like:
+
+ *	for {
+ *		selector := workflow.NewSelector(ctx)
+ *		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal1"), func(c workflow.ReceiveChannel, more bool) {
+ *			// Process signal1
+ *		})
+ *		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal2"), func(c workflow.ReceiveChannel, more bool) {
+ *			// Process signal2
+ *		}
+ *		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal3"), func(c workflow.ReceiveChannel, more bool) {
+ *			// Process signal3
+ *		}
+ *		cCtx, cancel := workflow.WithCancel(ctx)
+ *		timer := workflow.NewTimer(cCtx, timeToNextSignal)
+ *		selector.AddFuture(timer, func(f workflow.Future) {
+ *			// Process timeout
+ *		})
+ * 		selector.Select(ctx)
+ *	    cancel()
+ *      // break out of the loop on certain condition
+ *	}
+ *
+ *  The above implementation works. But it quickly becomes pretty convoluted if the number of signals
+ *  and rules around order of their arrivals and timeouts increases.
+ *
+ *  The following example demonstrates an alternative approach. It receives signals in a separate goroutine.
+ *  Each signal handler just updates a correspondent shared variable with the signal data.
+ *  The main workflow function awaits the next step using `workflow.AwaitWithTimeout` using condition composed of
+ *  the shared variables. This makes the main workflow method free from signal callbacks and makes the business logic
+ *  clear.
+ */
+
+// SignalToSignalTimeout is them maximum time between signals
+var SignalToSignalTimeout = 30 * time.Second
+
+// FromFirstSignalTimeout is the maximum time to receive all signals
+var FromFirstSignalTimeout = 60 * time.Second
+
+type AwaitSignals struct {
+	FirstSignalTime time.Time
+	Signal1Received bool
+	Signal2Received bool
+	Signal3Received bool
+}
+
+// Listen to signals Signal1, Signal2, and Signal3
+func (a *AwaitSignals) Listen(ctx workflow.Context) {
+	log := workflow.GetLogger(ctx)
+	for {
+		selector := workflow.NewSelector(ctx)
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal1"), func(c workflow.ReceiveChannel, more bool) {
+			c.Receive(ctx, nil)
+			a.Signal1Received = true
+			log.Info("Signal1 Received")
+		})
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal2"), func(c workflow.ReceiveChannel, more bool) {
+			c.Receive(ctx, nil)
+			a.Signal2Received = true
+			log.Info("Signal2 Received")
+		})
+		selector.AddReceive(workflow.GetSignalChannel(ctx, "Signal3"), func(c workflow.ReceiveChannel, more bool) {
+			c.Receive(ctx, nil)
+			a.Signal3Received = true
+			log.Info("Signal3 Received")
+		})
+		selector.Select(ctx)
+		if a.FirstSignalTime.IsZero() {
+			a.FirstSignalTime = workflow.Now(ctx)
+		}
+	}
+}
+
+// GetNextTimeout returns the maximum time allowed to wait for the next signal.
+func (a *AwaitSignals) GetNextTimeout(ctx workflow.Context) (time.Duration, error) {
+	if a.FirstSignalTime.IsZero() {
+		panic("FirstSignalTime is not yet set")
+	}
+	total := workflow.Now(ctx).Sub(a.FirstSignalTime)
+	totalLeft := FromFirstSignalTimeout - total
+	if totalLeft <= 0 {
+		return 0, temporal.NewApplicationError("FromFirstSignalTimeout", "timeout")
+	}
+	if SignalToSignalTimeout < totalLeft {
+		return SignalToSignalTimeout, nil
+	}
+	return totalLeft, nil
+}
+
+// AwaitSignalsWorkflow workflow definition
+func AwaitSignalsWorkflow(ctx workflow.Context) (err error) {
+	log := workflow.GetLogger(ctx)
+	var a AwaitSignals
+	// Listen to signals in a different goroutine
+	workflow.Go(ctx, a.Listen)
+
+	// Wait for Signal1
+	err = workflow.Await(ctx, func() bool {
+		return a.Signal1Received
+	})
+	// Cancellation
+	if err != nil {
+		return
+	}
+	log.Info("Signal1 Processed")
+
+	// Wait for Signal2
+	timeout, err := a.GetNextTimeout(ctx)
+	// No time left. At this point this cannot really happen.
+	if err != nil {
+		return
+	}
+	ok, err := workflow.AwaitWithTimeout(ctx, timeout, func() bool {
+		return a.Signal2Received
+	})
+	// Cancellation
+	if err != nil {
+		return
+	}
+	// timeout
+	if !ok {
+		return temporal.NewApplicationError("Timed out waiting for signal2", "timeout")
+	}
+	log.Info("Signal2 Processed")
+
+	// Wait for Signal3
+	timeout, err = a.GetNextTimeout(ctx)
+	// No time left.
+	if err != nil {
+		return
+	}
+	ok, err = workflow.AwaitWithTimeout(ctx, timeout, func() bool {
+		return a.Signal3Received
+	})
+	// Cancellation
+	if err != nil {
+		return
+	}
+	// timeout
+	if !ok {
+		return temporal.NewApplicationError("Timed out waiting for signal3", "timeout")
+	}
+	log.Info("Signal3 Processed")
+	return nil
+}

await-signals/await_signals_workflow_test.go

@@ -0,0 +1,61 @@
+package await_signals
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+	"go.temporal.io/sdk/testsuite"
+)
+
+type UnitTestSuite struct {
+	suite.Suite
+	testsuite.WorkflowTestSuite
+}
+
+func TestUnitTestSuite(t *testing.T) {
+	suite.Run(t, new(UnitTestSuite))
+}
+
+func (s *UnitTestSuite) Test_WorkflowTimeout() {
+	env := s.NewTestWorkflowEnvironment()
+	env.ExecuteWorkflow(AwaitSignalsWorkflow)
+
+	s.True(env.IsWorkflowCompleted())
+	// Workflow times out
+	s.Error(env.GetWorkflowError())
+}
+
+func (s *UnitTestSuite) Test_SignalsInOrder() {
+	env := s.NewTestWorkflowEnvironment()
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflow("Signal1", nil)
+	}, time.Hour)
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflow("Signal2", nil)
+	}, time.Hour+time.Second)
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflow("Signal3", nil)
+	}, time.Hour+3*time.Second)
+	env.ExecuteWorkflow(AwaitSignalsWorkflow)
+
+	s.True(env.IsWorkflowCompleted())
+	s.NoError(env.GetWorkflowError())
+}
+
+func (s *UnitTestSuite) Test_SignalsInReverseOrder() {
+	env := s.NewTestWorkflowEnvironment()
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflow("Signal3", nil)
+	}, time.Hour)
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflow("Signal2", nil)
+	}, time.Hour+time.Second)
+	env.RegisterDelayedCallback(func() {
+		env.SignalWorkflow("Signal1", nil)
+	}, time.Hour+3*time.Second)
+	env.ExecuteWorkflow(AwaitSignalsWorkflow)
+
+	s.True(env.IsWorkflowCompleted())
+	s.NoError(env.GetWorkflowError())
+}

await-signals/starter/main.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	await_signals "github.com/temporalio/samples-go/await-signals"
+	"log"
+	"math/rand"
+	"time"
+
+	"github.com/pborman/uuid"
+	"go.temporal.io/sdk/client"
+)
+
+func main() {
+	// The client is a heavyweight object that should be created once per process.
+	c, err := client.Dial(client.Options{
+		HostPort: client.DefaultHostPort,
+	})
+	if err != nil {
+		log.Fatalln("Unable to create client", err)
+	}
+	defer c.Close()
+
+	workflowOptions := client.StartWorkflowOptions{
+		ID:        "await_signals_" + uuid.New(),
+		TaskQueue: "await_signals",
+	}
+
+	we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, await_signals.AwaitSignalsWorkflow)
+	if err != nil {
+		log.Fatalln("Unable to execute workflow", err)
+	}
+	log.Println("Started workflow", "WorkflowID", we.GetID(), "RunID", we.GetRunID())
+
+	log.Println("Sending signals")
+	signals := []int{1, 2, 3}
+	// Send signals in random order
+	rand.Shuffle(len(signals), func(i, j int) { signals[i], signals[j] = signals[j], signals[i] })
+	for _, signal := range signals {
+		signalName := fmt.Sprintf("Signal%d", signal)
+		err = c.SignalWorkflow(context.Background(), we.GetID(), we.GetRunID(), signalName, nil)
+		if err != nil {
+			log.Fatalln("Unable to signals workflow", err)
+		}
+		log.Println("Sent " + signalName)
+		time.Sleep(2 * time.Second)
+	}
+
+}

await-signals/worker/main.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	await_signals "github.com/temporalio/samples-go/await-signals"
+	"log"
+
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+)
+
+func main() {
+	// The client and worker are heavyweight objects that should be created once per process.
+	c, err := client.Dial(client.Options{
+		HostPort: client.DefaultHostPort,
+	})
+	if err != nil {
+		log.Fatalln("Unable to create client", err)
+	}
+	defer c.Close()
+
+	w := worker.New(c, "await_signals", worker.Options{})
+
+	w.RegisterWorkflow(await_signals.AwaitSignalsWorkflow)
+
+	err = w.Run(worker.InterruptCh())
+	if err != nil {
+		log.Fatalln("Unable to start worker", err)
+	}
+}