Add Data Converter to support custom serialization/deserialization (#463)

* Add data converter to support custom encode/decode

* Add comments on DataConverter interface

* Address comments

* Address comments

Author: vancexu

.gitignore

@@ -16,3 +16,4 @@ test.log
 /dummy
 .build
 .bins
+.DS_Store

encoded/encoded.go

@@ -38,4 +38,22 @@ type (
 		// Get extract the encoded values into strong typed value pointers.
 		Get(valuePtr ...interface{}) error
 	}
+
+	// DataConverter is used by the framework to serialize/deserialize input and output of activity/workflow
+	// that need to be sent over the wire.
+	// To encode/decode workflow arguments, one should set DataConverter in two places:
+	//   1. Workflow worker, through worker.Options
+	//   2. Client, through client.Options
+	// To encode/decode Activity/ChildWorkflow arguments, one should set DataConverter in two places:
+	//   1. Inside workflow code, use workflow.WithDataConverter to create new Context,
+	// and pass that context to ExecuteActivity/ExecuteChildWorkflow calls.
+	// Cadence support using different DataConverters for different activity/childWorkflow in same workflow.
+	//   2. Activity/Workflow worker that run these activity/childWorkflow, through worker.Options.
+	DataConverter interface {
+		// ToData implements conversion of a list of values.
+		ToData(value ...interface{}) ([]byte, error)
+		// FromData implements conversion of an array of values of different types.
+		// Useful for deserializing arguments of function invocations.
+		FromData(input []byte, valuePtr ...interface{}) error
+	}
 )

internal/activity.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/uber-go/tally"
 	"go.uber.org/cadence/.gen/go/shared"
+	"go.uber.org/cadence/encoded"
 	"go.uber.org/cadence/internal/common"
 	"go.uber.org/zap"
 )
@@ -139,7 +140,7 @@ func RecordActivityHeartbeat(ctx context.Context, details ...interface{}) {
 	var err error
 	// We would like to be a able to pass in "nil" as part of details(that is no progress to report to)
 	if len(details) != 1 || details[0] != nil {
-		data, err = getHostEnvironment().encodeArgs(details)
+		data, err = encodeArgs(getDataConverterFromActivityCtx(ctx), details)
 		if err != nil {
 			panic(err)
 		}
@@ -168,6 +169,7 @@ func WithActivityTask(
 	invoker ServiceInvoker,
 	logger *zap.Logger,
 	scope tally.Scope,
+	dataConverter encoded.DataConverter,
 ) context.Context {
 	var deadline time.Time
 	scheduled := time.Unix(0, task.GetScheduledTimestamp())
@@ -198,6 +200,7 @@ func WithActivityTask(
 		scheduledTimestamp: scheduled,
 		startedTimestamp:   started,
 		taskList:           taskList,
+		dataConverter:      dataConverter,
 	})
 }
 

internal/activity_test.go

@@ -162,7 +162,7 @@ func (s *activityTestSuite) TestActivityHeartbeat_SuppressContinousInvokes() {
 	service3.EXPECT().RecordActivityTaskHeartbeat(gomock.Any(), gomock.Any(), callOptions...).
 		Return(&shared.RecordActivityTaskHeartbeatResponse{}, nil).
 		Do(func(ctx context.Context, request *shared.RecordActivityTaskHeartbeatRequest, opts ...yarpc.CallOption) {
-			ev := EncodedValues(request.Details)
+			ev := newEncodedValues(request.Details, nil)
 			var progress string
 			err := ev.Get(&progress)
 			if err != nil {

internal/client.go

@@ -241,8 +241,9 @@ type (
 
 	// ClientOptions are optional parameters for Client creation.
 	ClientOptions struct {
-		MetricsScope tally.Scope
-		Identity     string
+		MetricsScope  tally.Scope
+		Identity      string
+		DataConverter encoded.DataConverter
 	}
 
 	// StartWorkflowOptions configuration parameters for starting a workflow execution.
@@ -334,11 +335,18 @@ func NewClient(service workflowserviceclient.Interface, domain string, options *
 		metricScope = options.MetricsScope
 	}
 	metricScope = tagScope(metricScope, tagDomain, domain)
+	var dataConverter encoded.DataConverter
+	if options != nil && options.DataConverter != nil {
+		dataConverter = options.DataConverter
+	} else {
+		dataConverter = newDefaultDataConverter()
+	}
 	return &workflowClient{
 		workflowService: metrics.NewWorkflowServiceWrapper(service, metricScope),
 		domain:          domain,
 		metricsScope:    metrics.NewTaggedScope(metricScope),
 		identity:        identity,
+		dataConverter:   dataConverter,
 	}
 }
 
@@ -384,8 +392,7 @@ func (p WorkflowIDReusePolicy) toThriftPtr() *s.WorkflowIdReusePolicy {
 //   var result string // This need to be same type as the one passed to RecordHeartbeat
 //   NewValue(data).Get(&result)
 func NewValue(data []byte) encoded.Value {
-	result := EncodedValue(data)
-	return &result
+	return newEncodedValue(data, nil)
 }
 
 // NewValues creates a new encoded.Values which can be used to decode binary data returned by Cadence. For example:
@@ -396,6 +403,5 @@ func NewValue(data []byte) encoded.Value {
 //   var result2 int // These need to be same type as those arguments passed to RecordHeartbeat
 //   NewValues(data).Get(&result1, &result2)
 func NewValues(data []byte) encoded.Values {
-	result := EncodedValues(data)
-	return &result
+	return newEncodedValues(data, nil)
 }

internal/error.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 
 	"go.uber.org/cadence/.gen/go/shared"
+	"go.uber.org/cadence/encoded"
 )
 
 /*
@@ -87,7 +88,7 @@ type (
 	// CustomError returned from workflow and activity implementations with reason and optional details.
 	CustomError struct {
 		reason  string
-		details EncodedValues
+		details encoded.Values
 	}
 
 	// GenericError returned from workflow/workflow when the implementations return errors other than from NewCustomError() API.
@@ -98,12 +99,12 @@ type (
 	// TimeoutError returned when activity or child workflow timed out.
 	TimeoutError struct {
 		timeoutType shared.TimeoutType
-		details     EncodedValues
+		details     encoded.Values
 	}
 
 	// CanceledError returned when operation was canceled.
 	CanceledError struct {
-		details EncodedValues
+		details encoded.Values
 	}
 
 	// TerminatedError returned when workflow was terminated.
@@ -134,6 +135,9 @@ const (
 // ErrNoData is returned when trying to extract strong typed data while there is no data available.
 var ErrNoData = errors.New("no data available")
 
+// ErrTooManyArg is returned when trying to extract strong typed data with more arguments than available data.
+var ErrTooManyArg = errors.New("too many arguments")
+
 // ErrActivityResultPending is returned from activity's implementation to indicate the activity is not completed when
 // activity method returns. Activity needs to be completed by Client.CompleteActivity() separately. For example, if an
 // activity require human interaction (like approve an expense report), the activity could return activity.ErrResultPending
@@ -146,12 +150,14 @@ func NewCustomError(reason string, details ...interface{}) *CustomError {
 	if strings.HasPrefix(reason, "cadenceInternal:") {
 		panic("'cadenceInternal:' is reserved prefix, please use different reason")
 	}
-
-	data, err := getHostEnvironment().encodeArgs(details)
-	if err != nil {
-		panic(err)
+	// When return error to user, use EncodedValues as details and data is ready to be decoded by calling Get
+	if len(details) == 1 {
+		if d, ok := details[0].(*EncodedValues); ok {
+			return &CustomError{reason: reason, details: d}
+		}
 	}
-	return &CustomError{reason: reason, details: data}
+	// When create error for server, use ErrorDetailsValues as details to hold values and encode later
+	return &CustomError{reason: reason, details: ErrorDetailsValues(details)}
 }
 
 // NewTimeoutError creates TimeoutError instance.
@@ -162,20 +168,22 @@ func NewTimeoutError(timeoutType shared.TimeoutType) *TimeoutError {
 
 // NewHeartbeatTimeoutError creates TimeoutError instance
 func NewHeartbeatTimeoutError(details ...interface{}) *TimeoutError {
-	data, err := getHostEnvironment().encodeArgs(details)
-	if err != nil {
-		panic(err)
+	if len(details) == 1 {
+		if d, ok := details[0].(*EncodedValues); ok {
+			return &TimeoutError{timeoutType: shared.TimeoutTypeHeartbeat, details: d}
+		}
 	}
-	return &TimeoutError{timeoutType: shared.TimeoutTypeHeartbeat, details: data}
+	return &TimeoutError{timeoutType: shared.TimeoutTypeHeartbeat, details: ErrorDetailsValues(details)}
 }
 
 // NewCanceledError creates CanceledError instance
 func NewCanceledError(details ...interface{}) *CanceledError {
-	data, err := getHostEnvironment().encodeArgs(details)
-	if err != nil {
-		panic(err)
+	if len(details) == 1 {
+		if d, ok := details[0].(*EncodedValues); ok {
+			return &CanceledError{details: d}
+		}
 	}
-	return &CanceledError{details: data}
+	return &CanceledError{details: ErrorDetailsValues(details)}
 }
 
 // NewContinueAsNewError creates ContinueAsNewError instance
@@ -192,14 +200,14 @@ func NewCanceledError(details ...interface{}) *CanceledError {
 //
 func NewContinueAsNewError(ctx Context, wfn interface{}, args ...interface{}) *ContinueAsNewError {
 	// Validate type and its arguments.
-	workflowType, input, err := getValidatedWorkflowFunction(wfn, args)
-	if err != nil {
-		panic(err)
-	}
 	options := getWorkflowEnvOptions(ctx)
 	if options == nil {
 		panic("context is missing required options for continue as new")
 	}
+	workflowType, input, err := getValidatedWorkflowFunction(wfn, args, options.dataConverter)
+	if err != nil {
+		panic(err)
+	}
 	if options.taskListName == nil || *options.taskListName == "" {
 		panic("invalid task list provided")
 	}
@@ -230,11 +238,14 @@ func (e *CustomError) Reason() string {
 
 // HasDetails return if this error has strong typed detail data.
 func (e *CustomError) HasDetails() bool {
-	return e.details.HasValues()
+	return e.details != nil && e.details.HasValues()
 }
 
 // Details extracts strong typed detail data of this custom error. If there is no details, it will return ErrNoData.
 func (e *CustomError) Details(d ...interface{}) error {
+	if !e.HasDetails() {
+		return ErrNoData
+	}
 	return e.details.Get(d...)
 }
 
@@ -255,11 +266,14 @@ func (e *TimeoutError) TimeoutType() shared.TimeoutType {
 
 // HasDetails return if this error has strong typed detail data.
 func (e *TimeoutError) HasDetails() bool {
-	return e.details.HasValues()
+	return e.details != nil && e.details.HasValues()
 }
 
 // Details extracts strong typed detail data of this error. If there is no details, it will return ErrNoData.
 func (e *TimeoutError) Details(d ...interface{}) error {
+	if !e.HasDetails() {
+		return ErrNoData
+	}
 	return e.details.Get(d...)
 }
 
@@ -270,11 +284,14 @@ func (e *CanceledError) Error() string {
 
 // HasDetails return if this error has strong typed detail data.
 func (e *CanceledError) HasDetails() bool {
-	return e.details.HasValues()
+	return e.details != nil && e.details.HasValues()
 }
 
 // Details extracts strong typed detail data of this error.
 func (e *CanceledError) Details(d ...interface{}) error {
+	if !e.HasDetails() {
+		return ErrNoData
+	}
 	return e.details.Get(d...)
 }