feat: add cancel-in-progress sample

cancel-in-progress/README.md

@@ -0,0 +1,19 @@
+## Cancel-in-progress
+
+This example demonstrates how to implement a workflow that ensures that only one run of the child workflow is executed at a time. Subsequent runs will cancel the running child workflow and re-run it with the latest sent options through `SignalWithStartWorkflow`.
+Those semantics are useful, especially when implementing a CI pipeline. New commits during the workflow execution should short-circuit the child workflow and only build the most recent commit.
+
+
+Make sure the [Temporal Server is running locally](https://docs.temporal.io/docs/server/quick-install).
+
+From the root of the project, start a Worker:
+
+```bash
+go run cancel-in-progress/worker/main.go
+```
+
+Start the Workflow Execution:
+
+```bash
+go run cancel-in-progress/starter/main.go
+```

cancel-in-progress/child_workflow.go

@@ -0,0 +1,19 @@
+package cancel_in_progress
+
+import (
+ "go.temporal.io/sdk/workflow"
+ "time"
+)
+
+// SampleChildWorkflow is a Workflow Definition
+func SampleChildWorkflow(ctx workflow.Context, name string) (string, error) {
+ logger := workflow.GetLogger(ctx)
+
+ // Simulate some long running processing.
+ _ = workflow.Sleep(ctx, time.Second*3)
+
+ greeting := "Hello " + name + "!"
+ logger.Info("Child workflow execution: " + greeting)
+
+ return greeting, nil
+}

cancel-in-progress/parent_workflow.go

@@ -0,0 +1,109 @@
+package cancel_in_progress
+
+import (
+ "errors"
+ "go.temporal.io/sdk/log"
+ "go.temporal.io/sdk/workflow"
+)
+
+const ParentWorkflowSignalName = "parent-workflow-signal"
+
+func SampleParentWorkflow(ctx workflow.Context) (result string, err error) {
+ logger := workflow.GetLogger(ctx)
+
+ var message string
+
+ reBuildSignalChan := workflow.GetSignalChannel(ctx, ParentWorkflowSignalName)
+
+ // This will not block because the workflow is started with a signal transactional.
+ reBuildSignalChan.Receive(ctx, &message)
+
+ cwo := workflow.ChildWorkflowOptions{
+ WaitForCancellation: true,
+ }
+ ctx = workflow.WithChildOptions(ctx, cwo)
+
+ var isProcessingDone = false
+
+ for !isProcessingDone {
+ ctx, cancelHandler := workflow.WithCancel(ctx)
+
+ // it is important to re-execute the child workflow after every signal
+ // because we might have cancelled the previous execution
+ childWorkflowFuture := workflow.ExecuteChildWorkflow(ctx, SampleChildWorkflow, message)
+
+ selector := workflow.NewSelector(ctx)
+
+ selector.AddFuture(childWorkflowFuture, func(f workflow.Future) {
+ err = f.Get(ctx, &result)
+
+ // we don't want to end the parent workflow when child workflow is cancelled
+ if errors.Is(ctx.Err(), workflow.ErrCanceled) {
+ logger.Info("Child execution cancelled.")
+ return
+ }
+
+ if err != nil {
+ logger.Error("Child execution failure.", "Error", err)
+ }
+
+ // Child workflow completed. Let's end the parent workflow.
+ isProcessingDone = true
+ })
+
+ selector.AddReceive(reBuildSignalChan, func(c workflow.ReceiveChannel, more bool) {
+ logger.Info("Received signal.", "Message", message)
+
+ // cancel the child workflow as fast as possible after we received a new signal
+ // if a child workflow execution is in progress we cancel everything and start a new child workflow (for loop)
+ cancelHandler()
+
+ // drain the channel to get the latest signal
+ // Users might send multiple signals in a short period of time
+ // and we are only interested in the latest signal
+ message = GetLatestMessageFromChannel(logger, reBuildSignalChan)
+ })
+
+ // wait for the build workflow to finish or the signal to cancel the running execution through a signal
+ selector.Select(ctx)
+
+ // in case of an error we want to cancel the child workflow
+ if err != nil {
+ logger.Error("Child execution failure.", "Error", err)
+ return "", err
+ }
+ }
+
+ logger.Info("Parent execution completed.", "Result", result)
+
+ return result, nil
+}
+
+func GetLatestMessageFromChannel(logger log.Logger, ch workflow.ReceiveChannel) string {
+ var message string
+ var messages []string
+
+ for {
+ var m string
+ if ch.ReceiveAsync(&m) {
+ messages = append(messages, m)
+ logger.Info("Additional message received.", "message", message)
+ } else {
+ break
+ }
+ }
+
+ for i, m := range messages {
+ // continue with the latest signal
+ if i == len(messages)-1 {
+ // Update the workflow options to use the latest message for the next child workflow execution
+ message = m
+ logger.Info("Continue with latest message.", "message", message)
+ } else {
+ logger.Info("Cancel old workflow execution.", "message", message)
+ // You might want to do some cleanup here
+ }
+ }
+
+ return message
+}

cancel-in-progress/starter/main.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+ "context"
+ "go.temporal.io/sdk/client"
+ "log"
+ "math/rand"
+ "strconv"
+ "time"
+
+ cancel_in_progress "github.com/temporalio/samples-go/cancel-in-progress"
+)
+
+func main() {
+ // The client is a heavyweight object that should be created only 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()
+
+ // This Workflow ID must be fixed because we don't want to start a new workflow every time.
+ projectID := "my-project"
+
+ workflowID := "parent-workflow_" + projectID
+ workflowOptions := client.StartWorkflowOptions{
+ ID: workflowID,
+ TaskQueue: "child-workflow",
+ }
+
+ // Start the workflow execution or send a signal to an existing workflow execution.
+ workflowRun, err := c.SignalWithStartWorkflow(
+ context.Background(),
+ workflowID,
+ cancel_in_progress.ParentWorkflowSignalName,
+ "World",
+ workflowOptions,
+ cancel_in_progress.SampleParentWorkflow,
+ )
+ if err != nil {
+ log.Fatalln("Unable to execute workflow", err)
+ }
+ log.Println("Started workflow",
+ "WorkflowID", workflowRun.GetID(), "RunID", workflowRun.GetRunID())
+
+ // Send three signals to the workflow. At the end, we will expect that only the result of the last signal is returned.
+ for i := 1; i <= 3; i++ {
+ workflowRun, err = c.SignalWithStartWorkflow(
+ context.Background(),
+ workflowID,
+ cancel_in_progress.ParentWorkflowSignalName,
+ "World"+strconv.Itoa(i),
+ workflowOptions,
+ cancel_in_progress.SampleParentWorkflow,
+ )
+ if err != nil {
+ log.Fatalln("Unable to execute workflow", err)
+ }
+
+ time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
+
+ log.Println("Started workflow",
+ "WorkflowID", workflowRun.GetID(), "RunID", workflowRun.GetRunID())
+ }
+
+ // Synchronously wait for the Workflow Execution to complete.
+ // Behind the scenes the SDK performs a long poll operation.
+ // If you need to wait for the Workflow Execution to complete from another process use
+ // Client.GetWorkflow API to get an instance of the WorkflowRun.
+ var result string
+ err = workflowRun.Get(context.Background(), &result)
+ if err != nil {
+ log.Fatalln("Failure getting workflow result", err)
+ }
+ log.Printf("Workflow result: %v", result)
+}

cancel-in-progress/worker/main.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+ "log"
+
+ "go.temporal.io/sdk/client"
+ "go.temporal.io/sdk/worker"
+
+ cancel_in_progress "github.com/temporalio/samples-go/cancel-in-progress"
+)
+
+func main() {
+ // The client is a heavyweight object that should be created only 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, "child-workflow", worker.Options{})
+
+ w.RegisterWorkflow(cancel_in_progress.SampleParentWorkflow)
+ w.RegisterWorkflow(cancel_in_progress.SampleChildWorkflow)
+
+ err = w.Run(worker.InterruptCh())
+ if err != nil {
+ log.Fatalln("Unable to start worker", err)
+ }
+}