Merge pull request #163 from cschleiden/cschleiden/converter

Support passing in a custom converter for inputs/result serialization

Author: cschleiden

backend/backend.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 
+	"github.com/cschleiden/go-workflows/internal/converter"
 	core "github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/task"
@@ -70,4 +71,7 @@ type Backend interface {
 
 	// Metrics returns the configured metrics client for the backend
 	Metrics() metrics.Client
+
+	// Converter returns the configured converter for the backend
+	Converter() converter.Converter
 }

backend/mock_Backend.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
@@ -100,6 +101,10 @@ func (b *mysqlBackend) Metrics() metrics.Client {
 	return b.options.Metrics.WithTags(metrics.Tags{metrickeys.Backend: "mysql"})
 }
 
+func (b *mysqlBackend) Converter() converter.Converter {
+	return b.options.Converter
+}
+
 func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *workflow.Instance, event *history.Event) error {
 	tx, err := b.db.BeginTx(ctx, &sql.TxOptions{
 		Isolation: sql.LevelReadCommitted,

backend/mysql/mysql.go

@@ -3,6 +3,7 @@ package backend
 import (
 	"time"
 
+	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/logger"
 	mi "github.com/cschleiden/go-workflows/internal/metrics"
 	"github.com/cschleiden/go-workflows/log"
@@ -17,6 +18,10 @@ type Options struct {
 
 	TracerProvider trace.TracerProvider
 
+	// Converter is the converter to use for serializing and deserializing inputs and results. If not explicitly set
+	// converter.DefaultConverter is used.
+	Converter converter.Converter
+
 	StickyTimeout time.Duration
 
 	WorkflowLockTimeout time.Duration
@@ -32,6 +37,7 @@ var DefaultOptions Options = Options{
 	Logger:         logger.NewDefaultLogger(),
 	Metrics:        mi.NewNoopMetricsClient(),
 	TracerProvider: trace.NewNoopTracerProvider(),
+	Converter:      converter.DefaultConverter,
 }
 
 type BackendOption func(*Options)
@@ -60,6 +66,12 @@ func WithTracerProvider(tp trace.TracerProvider) BackendOption {
 	}
 }
 
+func WithConverter(converter converter.Converter) BackendOption {
+	return func(o *Options) {
+		o.Converter = converter
+	}
+}
+
 func ApplyOptions(opts ...BackendOption) Options {
 	options := DefaultOptions
 

backend/options.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
@@ -116,6 +117,10 @@ func (rb *redisBackend) Tracer() trace.Tracer {
 	return rb.options.TracerProvider.Tracer(backend.TracerName)
 }
 
+func (rb *redisBackend) Converter() converter.Converter {
+	return rb.options.Converter
+}
+
 func (rb *redisBackend) Close() error {
 	return rb.rdb.Close()
 }

backend/redis/redis.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
@@ -75,6 +76,10 @@ func (sb *sqliteBackend) Tracer() trace.Tracer {
 	return sb.options.TracerProvider.Tracer(backend.TracerName)
 }
 
+func (sb *sqliteBackend) Converter() converter.Converter {
+	return sb.options.Converter
+}
+
 func (sb *sqliteBackend) CreateWorkflowInstance(ctx context.Context, instance *workflow.Instance, event history.Event) error {
 	tx, err := sb.db.BeginTx(ctx, nil)
 	if err != nil {

backend/sqlite/sqlite.go

@@ -10,7 +10,6 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	"github.com/cschleiden/go-workflows/backend"
 	a "github.com/cschleiden/go-workflows/internal/args"
-	"github.com/cschleiden/go-workflows/internal/converter"
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/fn"
 	"github.com/cschleiden/go-workflows/internal/history"
@@ -56,7 +55,7 @@ func New(backend backend.Backend) Client {
 }
 
 func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowInstanceOptions, wf workflow.Workflow, args ...interface{}) (*workflow.Instance, error) {
-	inputs, err := a.ArgsToInputs(converter.DefaultConverter, args...)
+	inputs, err := a.ArgsToInputs(c.backend.Converter(), args...)
 	if err != nil {
 		return nil, fmt.Errorf("converting arguments: %w", err)
 	}
@@ -101,7 +100,7 @@ func (c *client) CancelWorkflowInstance(ctx context.Context, instance *workflow.
 }
 
 func (c *client) SignalWorkflow(ctx context.Context, instanceID string, name string, arg interface{}) error {
-	input, err := converter.DefaultConverter.To(arg)
+	input, err := c.backend.Converter().To(arg)
 	if err != nil {
 		return fmt.Errorf("converting arguments: %w", err)
 	}
@@ -184,7 +183,7 @@ func GetWorkflowResult[T any](ctx context.Context, c Client, instance *workflow.
 			}
 
 			var r T
-			if err := converter.DefaultConverter.From(a.Result, &r); err != nil {
+			if err := b.Converter().From(a.Result, &r); err != nil {
 				return *new(T), fmt.Errorf("converting result: %w", err)
 			}
 

client/client.go

@@ -58,6 +58,7 @@ func Test_Client_GetWorkflowResultSuccess(t *testing.T) {
 			Error:  "",
 		}),
 	}, nil)
+	b.On("Converter").Return(converter.DefaultConverter)
 
 	c := &client{
 		backend: b,
@@ -77,6 +78,7 @@ func Test_Client_SignalWorkflow(t *testing.T) {
 
 	b := &backend.MockBackend{}
 	b.On("Logger").Return(logger.NewDefaultLogger())
+	b.On("Converter").Return(converter.DefaultConverter)
 	b.On("SignalWorkflow", ctx, instanceID, mock.MatchedBy(func(event history.Event) bool {
 		return event.Type == history.EventType_SignalReceived &&
 			event.Attributes.(*history.SignalReceivedAttributes).Name == "test"
@@ -104,6 +106,7 @@ func Test_Client_SignalWorkflow_WithArgs(t *testing.T) {
 
 	b := &backend.MockBackend{}
 	b.On("Logger").Return(logger.NewDefaultLogger())
+	b.On("Converter").Return(converter.DefaultConverter)
 	b.On("SignalWorkflow", ctx, instanceID, mock.MatchedBy(func(event history.Event) bool {
 		return event.Type == history.EventType_SignalReceived &&
 			event.Attributes.(*history.SignalReceivedAttributes).Name == "test" &&

client/client_test.go

@@ -19,16 +19,18 @@ import (
 )
 
 type Executor struct {
-	logger log.Logger
-	tracer trace.Tracer
-	r      *workflow.Registry
+	logger    log.Logger
+	tracer    trace.Tracer
+	converter converter.Converter
+	r         *workflow.Registry
 }
 
-func NewExecutor(logger log.Logger, tracer trace.Tracer, r *workflow.Registry) Executor {
+func NewExecutor(logger log.Logger, tracer trace.Tracer, converter converter.Converter, r *workflow.Registry) Executor {
 	return Executor{
-		logger: logger,
-		tracer: tracer,
-		r:      r,
+		logger:    logger,
+		tracer:    tracer,
+		converter: converter,
+		r:         r,
 	}
 }
 
@@ -45,7 +47,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *task.Activity) (pa
 		return nil, errors.New("activity not a function")
 	}
 
-	args, addContext, err := args.InputsToArgs(converter.DefaultConverter, activityFn, a.Inputs)
+	args, addContext, err := args.InputsToArgs(e.converter, activityFn, a.Inputs)
 	if err != nil {
 		return nil, fmt.Errorf("converting activity inputs: %w", err)
 	}
@@ -79,7 +81,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *task.Activity) (pa
 
 	if len(r) > 1 {
 		var err error
-		result, err = converter.DefaultConverter.To(r[0].Interface())
+		result, err = e.converter.To(r[0].Interface())
 		if err != nil {
 			return nil, fmt.Errorf("converting activity result: %w", err)
 		}

internal/activity/executor.go

@@ -0,0 +1,13 @@
+package converter
+
+import "github.com/cschleiden/go-workflows/internal/sync"
+
+type converterKey struct{}
+
+func WithConverter(ctx sync.Context, converter Converter) sync.Context {
+	return sync.WithValue(ctx, converterKey{}, converter)
+}
+
+func GetConverter(ctx sync.Context) Converter {
+	return ctx.Value(converterKey{}).(Converter)
+}