Partial sub-workflow support

Author: cschleiden

backend/redis/instance.go

@@ -14,14 +14,8 @@ import (
 )
 
 func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, event history.WorkflowEvent) error {
-	// Store instance with its state
-	if err := createInstance(ctx, rb.rdb, event.WorkflowInstance, &instanceState{
-		InstanceID:  event.WorkflowInstance.GetInstanceID(),
-		ExecutionID: event.WorkflowInstance.GetExecutionID(),
-		State:       backend.WorkflowStateActive,
-		CreatedAt:   time.Now(),
-	}); err != nil {
-		return errors.Wrap(err, "could not create workflow instance")
+	if err := createInstance(ctx, rb.rdb, event.WorkflowInstance, false); err != nil {
+		return err
 	}
 
 	// Create event stream
@@ -91,10 +85,15 @@ type instanceState struct {
 	CompletedAt *time.Time            `json:"completed_at,omitempty"`
 }
 
-func createInstance(ctx context.Context, rdb redis.UniversalClient, instance core.WorkflowInstance, state *instanceState) error {
+func createInstance(ctx context.Context, rdb redis.UniversalClient, instance core.WorkflowInstance, ignoreDuplicate bool) error {
 	key := instanceKey(instance.GetInstanceID())
 
-	b, err := json.Marshal(state)
+	b, err := json.Marshal(&instanceState{
+		InstanceID:  instance.GetInstanceID(),
+		ExecutionID: instance.GetExecutionID(),
+		State:       backend.WorkflowStateActive,
+		CreatedAt:   time.Now(),
+	})
 	if err != nil {
 		return errors.Wrap(err, "could not marshal instance state")
 	}
@@ -104,7 +103,7 @@ func createInstance(ctx context.Context, rdb redis.UniversalClient, instance cor
 		return errors.Wrap(err, "could not store instance")
 	}
 
-	if !ok {
+	if !ignoreDuplicate && !ok {
 		return errors.New("workflow instance already exists")
 	}
 

backend/redis/workflow.go

@@ -103,13 +103,14 @@ func (rb *redisBackend) CompleteWorkflowTask(ctx context.Context, taskID string,
 	for targetInstance, events := range groupedEvents {
 		if instance.GetInstanceID() != targetInstance.GetInstanceID() {
 			// Create new instance
-			// TODO: Support creating sub-workflows
-			panic("not implemented")
+			if err := createInstance(ctx, rb.rdb, targetInstance, true); err != nil {
+				return err
+			}
 		}
 
 		// Insert pending events for target instance
 		for _, event := range events {
-			if err := addEventToStream(ctx, rb.rdb, pendingEventsKey(instance.GetInstanceID()), &event); err != nil {
+			if err := addEventToStream(ctx, rb.rdb, pendingEventsKey(targetInstance.GetInstanceID()), &event); err != nil {
 				return err
 			}
 		}

samples/signal/signal.go

@@ -101,7 +101,7 @@ func Workflow1(ctx workflow.Context, msg string, subID string) (string, error) {
 func SubWorkflow1(ctx workflow.Context) (string, error) {
 	samples.Trace(ctx, "Waiting for signal from sub-worflow")
 
-	c := workflow.NewSignalChannel[string](ctx, "sub-signal")
+	c := workflow.NewSignalChannel[int](ctx, "sub-signal")
 	c.Receive(ctx)
 
 	samples.Trace(ctx, "Received sub-workflow signal")

samples/subworkflow/subworkflow.go

@@ -7,8 +7,9 @@ import (
 	"os/signal"
 
 	"github.com/cschleiden/go-workflows/backend"
-	"github.com/cschleiden/go-workflows/backend/sqlite"
+	"github.com/cschleiden/go-workflows/backend/redis"
 	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/samples"
 	"github.com/cschleiden/go-workflows/worker"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/google/uuid"
@@ -18,8 +19,12 @@ import (
 func main() {
 	ctx := context.Background()
 
-	b := sqlite.NewInMemoryBackend()
+	// b := sqlite.NewInMemoryBackend()
 	// b := sqlite.NewSqliteBackend("subworkflow.sqlite")
+	b, err := redis.NewRedisBackend("localhost:6379", "", "RedisPassw0rd", 0)
+	if err != nil {
+		panic(err)
+	}
 
 	// Run worker
 	go RunWorker(ctx, b)
@@ -60,66 +65,48 @@ func RunWorker(ctx context.Context, mb backend.Backend) {
 }
 
 func Workflow1(ctx workflow.Context, msg string) error {
-	log.Println("Entering Workflow1")
-	log.Println("\tWorkflow instance input:", msg)
-	log.Println("\tIsReplaying:", workflow.Replaying(ctx))
-
-	defer func() {
-		log.Println("Leaving Workflow1")
-	}()
+	samples.Trace(ctx, "Entering Workflow1")
+	samples.Trace(ctx, "\tWorkflow instance input:", msg)
 
 	wr, err := workflow.CreateSubWorkflowInstance[string](ctx, workflow.DefaultSubWorkflowOptions, Workflow2, "some input").Get(ctx)
 	if err != nil {
 		return errors.Wrap(err, "could not get sub workflow result")
 	}
 
-	log.Println("Sub workflow result:", wr)
+	samples.Trace(ctx, "Sub workflow result:", wr)
 
 	return nil
 }
 
 func Workflow2(ctx workflow.Context, msg string) (string, error) {
-	log.Println("Entering Workflow2")
-	log.Println("\tWorkflow instance input:", msg)
-	log.Println("\tIsReplaying:", workflow.Replaying(ctx))
-
-	defer func() {
-		log.Println("Leaving Workflow2")
-	}()
+	samples.Trace(ctx, "Entering Workflow2")
+	samples.Trace(ctx, "\tWorkflow instance input:", msg)
 
 	r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1, 35, 12).Get(ctx)
 	if err != nil {
 		panic("error getting activity 1 result")
 	}
-	log.Println("R1 result:", r1)
-	log.Println("\tIsReplaying:", workflow.Replaying(ctx))
+	samples.Trace(ctx, "R1 result:", r1)
 
 	r2, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity2).Get(ctx)
 	if err != nil {
 		panic("error getting activity 1 result")
 	}
-	log.Println("R2 result:", r2)
-	log.Println("\tIsReplaying:", workflow.Replaying(ctx))
+	samples.Trace(ctx, "R2 result:", r2)
 
 	return "W2 Result", nil
 }
 
 func Activity1(ctx context.Context, a, b int) (int, error) {
 	log.Println("Entering Activity1")
-
-	defer func() {
-		log.Println("Leaving Activity1")
-	}()
+	defer log.Println("Leaving Activity1")
 
 	return a + b, nil
 }
 
 func Activity2(ctx context.Context) (int, error) {
 	log.Println("Entering Activity2")
-
-	defer func() {
-		log.Println("Leaving Activity2")
-	}()
+	defer log.Println("Leaving Activity2")
 
 	return 12, nil
 }