Support for signals

Author: cschleiden

backend/redis/instance.go

@@ -30,19 +30,18 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, event histor
 		return err
 	}
 
-	cmd := rb.rdb.XAdd(ctx, &redis.XAddArgs{
+	_, err = rb.rdb.XAdd(ctx, &redis.XAddArgs{
 		Stream: pendingEventsKey(event.WorkflowInstance.GetInstanceID()),
 		ID:     "*",
 		Values: map[string]interface{}{
 			"event": string(eventData),
 		},
-	})
-	_, err = cmd.Result()
+	}).Result()
 	if err != nil {
 		return errors.Wrap(err, "could not create event stream")
 	}
 
-	// Add instance to pending instances set
+	// Queue workflow instance task
 	if _, err := rb.workflowQueue.Enqueue(ctx, event.WorkflowInstance.GetInstanceID(), nil); err != nil {
 		if err != taskqueue.ErrTaskAlreadyInQueue {
 			return errors.Wrap(err, "could not queue workflow task")
@@ -53,7 +52,22 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, event histor
 }
 
 func (rb *redisBackend) GetWorkflowInstanceHistory(ctx context.Context, instance core.WorkflowInstance) ([]history.Event, error) {
-	panic("unimplemented")
+	msgs, err := rb.rdb.XRange(ctx, historyKey(instance.GetInstanceID()), "-", "+").Result()
+	if err != nil {
+		return nil, err
+	}
+
+	var events []history.Event
+	for _, msg := range msgs {
+		var event history.Event
+		if err := json.Unmarshal([]byte(msg.Values["event"].(string)), &event); err != nil {
+			return nil, errors.Wrap(err, "could not unmarshal event")
+		}
+
+		events = append(events, event)
+	}
+
+	return events, nil
 }
 
 func (rb *redisBackend) GetWorkflowInstanceState(ctx context.Context, instance core.WorkflowInstance) (backend.WorkflowState, error) {

backend/redis/signal.go

@@ -3,12 +3,21 @@ package redis
 import (
 	"context"
 
+	"github.com/cschleiden/go-workflows/backend/redis/taskqueue"
 	"github.com/cschleiden/go-workflows/internal/history"
+	"github.com/pkg/errors"
 )
 
 func (rb *redisBackend) SignalWorkflow(ctx context.Context, instanceID string, event history.Event) error {
-	// TODO: Store signal event
-	// TODO: Queue workflow task
+	if err := addEventToStream(ctx, rb.rdb, pendingEventsKey(instanceID), &event); err != nil {
+		return errors.Wrap(err, "could not add event to stream")
+	}
 
-	panic("unimplemented")
+	if _, err := rb.workflowQueue.Enqueue(ctx, instanceID, nil); err != nil {
+		if err != taskqueue.ErrTaskAlreadyInQueue {
+			return errors.Wrap(err, "could not queue workflow task")
+		}
+	}
+
+	return nil
 }

backend/redis/taskqueue/queue.go

@@ -3,7 +3,6 @@ package taskqueue
 import (
 	"context"
 	"encoding/json"
-	"log"
 	"time"
 
 	"github.com/go-redis/redis/v8"
@@ -170,8 +169,6 @@ func (q *taskQueue[T]) Complete(ctx context.Context, taskID string) error {
 		return errors.New("could find task to complete")
 	}
 
-	log.Println("Completing activity task", taskID)
-
 	return nil
 }
 

backend/redis/workflow.go

@@ -153,8 +153,12 @@ func (rb *redisBackend) CompleteWorkflowTask(ctx context.Context, taskID string,
 	}
 
 	// If there are pending events, enqueue the instance again
-	// TODO: Check for pending events
-	if state != backend.WorkflowStateFinished {
+	pendingCount, err := rb.rdb.XLen(ctx, pendingEventsKey(instance.GetInstanceID())).Result()
+	if err != nil {
+		return errors.Wrap(err, "could not read event stream")
+	}
+
+	if state != backend.WorkflowStateFinished && pendingCount > 0 {
 		if _, err := rb.workflowQueue.Enqueue(ctx, instance.GetInstanceID(), nil); err != nil {
 			if err != taskqueue.ErrTaskAlreadyInQueue {
 				return errors.Wrap(err, "could not queue workflow")