chore: support custom payload conveters

Signed-off-by: Valery Piashchynski <[email protected]>

Author: rustatian

internal.go

@@ -38,8 +38,19 @@ func (p *Plugin) initPool() error {
 		return err
 	}
 
-	dc := dataconverter.NewDataConverter(converter.GetDefaultDataConverter())
-	codec := proto.NewCodec(p.log, dc)
+	dc := make([]converter.PayloadConverter, 0, 5)
+	// standard payload converters
+	dc = append(dc, converter.NewNilPayloadConverter())
+	dc = append(dc, converter.NewByteSlicePayloadConverter())
+	dc = append(dc, converter.NewProtoJSONPayloadConverter())
+	dc = append(dc, converter.NewProtoPayloadConverter())
+
+	if p.temporal.customDataConverter != nil {
+		dc = append(dc, p.temporal.customDataConverter)
+	}
+
+	rrdc := dataconverter.NewDataConverter(converter.NewCompositeDataConverter(dc...))
+	codec := proto.NewCodec(p.log, rrdc)
 
 	// LA + A definitions
 	actDef := aggregatedpool.NewActivityDefinition(codec, ap, p.log, p.config.DisableActivityWorkers)
@@ -85,7 +96,7 @@ func (p *Plugin) initPool() error {
 		return errors.Str("worker info should contain at least 1 worker")
 	}
 
-	err = p.initTemporalClient(wi[0].PhpSdkVersion, wi[0].Flags, dc)
+	err = p.initTemporalClient(wi[0].PhpSdkVersion, wi[0].Flags, rrdc)
 	if err != nil {
 		return err
 	}

plugin.go

@@ -20,6 +20,7 @@ import (
 	"github.com/temporalio/roadrunner-temporal/v5/internal"
 	"github.com/temporalio/roadrunner-temporal/v5/internal/codec/proto"
 	tclient "go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/converter"
 	"go.temporal.io/sdk/worker"
 	"go.uber.org/zap"
 
@@ -61,7 +62,8 @@ type temporal struct {
 	client        tclient.Client
 	workers       []worker.Worker
 
-	interceptors map[string]api.Interceptor
+	interceptors        map[string]api.Interceptor
+	customDataConverter converter.PayloadConverter
 }
 
 type Plugin struct {
@@ -285,7 +287,7 @@ func (p *Plugin) Workers() []*process.State {
 	for i := range wfPw {
 		st, err := process.WorkerProcessState(wfPw[i])
 		if err != nil {
-			// log error and continue
+			// log the error and continue
 			p.log.Error("worker process state error", zap.Error(err))
 			continue
 		}
@@ -296,7 +298,7 @@ func (p *Plugin) Workers() []*process.State {
 	for i := range actPw {
 		st, err := process.WorkerProcessState(actPw[i])
 		if err != nil {
-			// log error and continue
+			// log the error and continue
 			p.log.Error("worker process state error", zap.Error(err))
 			continue
 		}
@@ -410,6 +412,9 @@ func (p *Plugin) Collects() []*dep.In {
 			p.temporal.interceptors[mdw.Name()] = mdw
 			p.mu.Unlock()
 		}, (*api.Interceptor)(nil)),
+		dep.Fits(func(pp any) {
+			p.temporal.customDataConverter = pp.(converter.PayloadConverter)
+		}, (*converter.PayloadConverter)(nil)),
 	}
 }