Add Decoder API in client (#435)

* fix some typo

* Add Decoder API in client

* remove unused god encoding

* use NewValue NewValues instead of new API

* remove unused code

* make fmt

* add example to comments

Author: vancexu

activity/activity.go

@@ -93,7 +93,7 @@ func GetMetricsScope(ctx context.Context) tally.Scope {
 //  TODO: we don't have a way to distinguish between the two cases when context is cancelled because
 //  context doesn't support overriding value of ctx.Error.
 //  TODO: Implement automatic heartbeating with cancellation through ctx.
-// details - the details that you provided here can be seen in the worflow when it receives TimeoutError, you
+// details - the details that you provided here can be seen in the workflow when it receives TimeoutError, you
 // can check error TimeOutType()/Details().
 func RecordHeartbeat(ctx context.Context, details ...interface{}) {
 	internal.RecordActivityHeartbeat(ctx, details...)

client/client.go

@@ -308,3 +308,24 @@ var _ Client = internal.Client(nil)
 var _ internal.Client = Client(nil)
 var _ DomainClient = internal.DomainClient(nil)
 var _ internal.DomainClient = DomainClient(nil)
+
+// NewValue creates a new encoded.Value which can be used to decode binary data returned by Cadence.  For example:
+// User had Activity.RecordHeartbeat(ctx, "my-heartbeat") and then got response from calling Client.DescribeWorkflowExecution.
+// The response contains binary field PendingActivityInfo.HeartbeatDetails,
+// which can be decoded by using:
+//   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 {
+	return internal.NewValue(data)
+}
+
+// NewValues creates a new encoded.Values which can be used to decode binary data returned by Cadence. For example:
+// User had Activity.RecordHeartbeat(ctx, "my-heartbeat", 123) and then got response from calling Client.DescribeWorkflowExecution.
+// The response contains binary field PendingActivityInfo.HeartbeatDetails,
+// which can be decoded by using:
+//   var result1 string
+//   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 {
+	return internal.NewValues(data)
+}

internal/client.go

@@ -377,3 +377,26 @@ func (p WorkflowIDReusePolicy) toThriftPtr() *s.WorkflowIdReusePolicy {
 	}
 	return &policy
 }
+
+// NewValue creates a new encoded.Value which can be used to decode binary data returned by Cadence.  For example:
+// User had Activity.RecordHeartbeat(ctx, "my-heartbeat") and then got response from calling Client.DescribeWorkflowExecution.
+// The response contains binary field PendingActivityInfo.HeartbeatDetails,
+// which can be decoded by using:
+//   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
+}
+
+// NewValues creates a new encoded.Values which can be used to decode binary data returned by Cadence. For example:
+// User had Activity.RecordHeartbeat(ctx, "my-heartbeat", 123) and then got response from calling Client.DescribeWorkflowExecution.
+// The response contains binary field PendingActivityInfo.HeartbeatDetails,
+// which can be decoded by using:
+//   var result1 string
+//   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
+}

internal/context.go

@@ -29,7 +29,7 @@ import (
 
 // Context is a clone of context.Context with Done() returning Channel instead
 // of native channel.
-// A Context carries a deadline, a cancelation signal, and other values across
+// A Context carries a deadline, a cancellation signal, and other values across
 // API boundaries.
 //
 // Context's methods may be called by multiple goroutines simultaneously.
@@ -69,7 +69,7 @@ type Context interface {
 	//  }
 	//
 	// See http://blog.golang.org/pipelines for more examples of how to use
-	// a Done channel for cancelation.
+	// a Done channel for cancellation.
 	Done() Channel
 
 	// Err returns a non-nil error value after Done is closed.  Err returns

internal/internal_utils_test.go

@@ -34,3 +34,41 @@ func TestChannelBuilderOptions(t *testing.T) {
 
 	require.Equal(t, time.Minute, builder.Timeout)
 }
+
+type testDecodeStruct struct {
+	Name string
+	Age  int
+}
+
+func TestNewValues(t *testing.T) {
+	var details []interface{}
+	heartbeatDetail := "status-report-to-workflow"
+	heartbeatDetail2 := 1
+	heartbeatDetail3 := testDecodeStruct{
+		Name: heartbeatDetail,
+		Age:  heartbeatDetail2,
+	}
+	details = append(details, heartbeatDetail, heartbeatDetail2, heartbeatDetail3)
+	data, err := getHostEnvironment().encodeArgs(details)
+	if err != nil {
+		panic(err)
+	}
+	var res string
+	var res2 int
+	var res3 testDecodeStruct
+	NewValues(data).Get(&res, &res2, &res3)
+	require.Equal(t, heartbeatDetail, res)
+	require.Equal(t, heartbeatDetail2, res2)
+	require.Equal(t, heartbeatDetail3, res3)
+}
+
+func TestNewValue(t *testing.T) {
+	heartbeatDetail := "status-report-to-workflow"
+	data, err := getHostEnvironment().encodeArg(heartbeatDetail)
+	if err != nil {
+		panic(err)
+	}
+	var res string
+	NewValue(data).Get(&res)
+	require.Equal(t, res, heartbeatDetail)
+}

internal/internal_worker.go

@@ -25,7 +25,6 @@ package internal
 import (
 	"bytes"
 	"context"
-	"encoding/gob"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -1140,43 +1139,6 @@ type encoding interface {
 	Unmarshal([]byte, []interface{}) error
 }
 
-// gobEncoding encapsulates gob encoding and decoding
-type gobEncoding struct {
-}
-
-// Register implements the encoding interface
-func (g gobEncoding) Register(obj interface{}) error {
-	gob.Register(obj)
-	return nil
-}
-
-// Marshal encodes an array of object into bytes
-func (g gobEncoding) Marshal(objs []interface{}) ([]byte, error) {
-	var buf bytes.Buffer
-	enc := gob.NewEncoder(&buf)
-	for i, obj := range objs {
-		if err := enc.Encode(obj); err != nil {
-			return nil, fmt.Errorf(
-				"unable to encode argument: %d, %v, with gob error: %v", i, reflect.TypeOf(obj), err)
-		}
-	}
-	return buf.Bytes(), nil
-}
-
-// Unmarshal decodes a byte array into the passed in objects
-// TODO: To deal with different number of arguments, may be encode number of arguments as a first value as well.
-// so we can decode if a ssubset of them are asked.
-func (g gobEncoding) Unmarshal(data []byte, objs []interface{}) error {
-	dec := gob.NewDecoder(bytes.NewBuffer(data))
-	for i, obj := range objs {
-		if err := dec.Decode(obj); err != nil {
-			return fmt.Errorf(
-				"unable to decode argument: %d, %v, with gob error: %v", i, reflect.TypeOf(obj), err)
-		}
-	}
-	return nil
-}
-
 // jsonEncoding encapsulates json encoding and decoding
 type jsonEncoding struct {
 }

internal/internal_worker_test.go

@@ -841,14 +841,6 @@ type encodingTest struct {
 	input    []interface{}
 }
 
-var encodingTests = []encodingTest{
-	{&gobEncoding{}, []interface{}{"test"}},
-	{&gobEncoding{}, []interface{}{"test1", "test2"}},
-	{&gobEncoding{}, []interface{}{"test1", 1, false}},
-	{&gobEncoding{}, []interface{}{"test1", testWorkflowResult{V: 10}, false}},
-	{&gobEncoding{}, []interface{}{"test2", &testWorkflowResult{V: 20}, 4}},
-}
-
 // duplicate of GetHostEnvironment().registerType()
 func testRegisterType(enc encoding, v interface{}) error {
 	t := reflect.Indirect(reflect.ValueOf(v)).Type()
@@ -859,30 +851,6 @@ func testRegisterType(enc encoding, v interface{}) error {
 	return enc.Register(arg)
 }
 
-func TestGobEncoding(t *testing.T) {
-	for _, et := range encodingTests {
-		for _, obj := range et.input {
-			err := testRegisterType(et.encoding, obj)
-			require.NoError(t, err, err)
-		}
-		data, err := et.encoding.Marshal(et.input)
-		require.NoError(t, err, err)
-
-		var result []interface{}
-		for _, v := range et.input {
-			arg := reflect.New(reflect.TypeOf(v)).Interface()
-			result = append(result, arg)
-		}
-		err = et.encoding.Unmarshal(data, result)
-		require.NoError(t, err, err)
-
-		for i := 0; i < len(et.input); i++ {
-			vat := reflect.ValueOf(result[i]).Elem().Interface()
-			require.Equal(t, et.input[i], vat)
-		}
-	}
-}
-
 func Test_ActivityNilArgs(t *testing.T) {
 	nilErr := errors.New("nils")
 	activityFn := func(name string, idx int, strptr *string) error {

workflow/context.go

@@ -26,7 +26,7 @@ import (
 
 // Context is a clone of context.Context with Done() returning Channel instead
 // of native channel.
-// A Context carries a deadline, a cancelation signal, and other values across
+// A Context carries a deadline, a cancellation signal, and other values across
 // API boundaries.
 //
 // Context's methods may be called by multiple goroutines simultaneously.

workflow/workflow.go

@@ -161,7 +161,7 @@ func ExecuteLocalActivity(ctx Context, activity interface{}, args ...interface{}
 // ExecuteChildWorkflow requests child workflow execution in the context of a workflow.
 // Context can be used to pass the settings for the child workflow.
 // For example: task list that this child workflow should be routed, timeouts that need to be configured.
-// Use ChildWorkflowOptions to pass down th`e options.
+// Use ChildWorkflowOptions to pass down the options.
 //  cwo := ChildWorkflowOptions{
 // 	    ExecutionStartToCloseTimeout: 10 * time.Minute,
 // 	    TaskStartToCloseTimeout: time.Minute,