Skip to content

Commit 5f7230d

Browse files
alexshtinalexshtin
authored
Use Payload for SearchAttributes, Memo, and Headers (#117)
* Use Payload for SA, Memo, and Headers. * Use Payload for SA, Memo, and Headers. * Fix check.
1 parent 6bcd477 commit 5f7230d

21 files changed

+171
-142
lines changed

‎encoded/encoded.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,17 @@ type (
4646
// Temporal support using different DataConverters for different activity/childWorkflow in same workflow.
4747
// 2. Activity/Workflow worker that run these activity/childWorkflow, through worker.Options.
4848
DataConverter = internal.DataConverter
49+
50+
// PayloadConverter converts single value to/from payload.
51+
PayloadConverter = internal.PayloadConverter
4952
)
5053

5154
// GetDefaultDataConverter return default data converter used by Temporal worker
5255
func GetDefaultDataConverter() DataConverter {
5356
return internal.DefaultDataConverter
5457
}
58+
59+
// GetDefaultPayloadConverter return default data converter used by Temporal worker
60+
func GetDefaultPayloadConverter() PayloadConverter {
61+
return internal.DefaultPayloadConverter
62+
}

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ require (
2222
github.com/uber-go/tally v3.3.15+incompatible
2323
github.com/uber/jaeger-client-go v2.22.1+incompatible
2424
github.com/uber/jaeger-lib v2.2.0+incompatible // indirect
25-
go.temporal.io/temporal-proto v0.20.30
25+
go.temporal.io/temporal-proto v0.20.31
2626
go.uber.org/atomic v1.6.0
2727
go.uber.org/goleak v1.0.0
2828
go.uber.org/zap v1.14.1

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,8 +93,8 @@ github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMW
9393
github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=
9494
github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
9595
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
96-
go.temporal.io/temporal-proto v0.20.30 h1:QxvCfTZ1U686bmlMPTTg0F/dvMvt02m7i3jF3zWEX/E=
97-
go.temporal.io/temporal-proto v0.20.30/go.mod h1:Lv8L8YBpbp0Z7V5nbvw5UD0j7x0isebhCOIDLkBqn6s=
96+
go.temporal.io/temporal-proto v0.20.31 h1:AlY49UhslnoUSV9HvnEewgy0ursxMPrOJAaQZHmDwzM=
97+
go.temporal.io/temporal-proto v0.20.31/go.mod h1:Lv8L8YBpbp0Z7V5nbvw5UD0j7x0isebhCOIDLkBqn6s=
9898
go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk=
9999
go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
100100
go.uber.org/goleak v1.0.0 h1:qsup4IcBdlmsnGfqyLl4Ntn3C2XCCuKAE7DwHpScyUo=

‎internal/encoded.go‎

Lines changed: 82 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -71,22 +71,32 @@ type (
7171
ToData(value ...interface{}) (*commonpb.Payloads, error)
7272
// FromData implements conversion of an array of values of different types.
7373
// Useful for deserializing arguments of function invocations.
74-
FromData(input *commonpb.Payloads, valuePtr ...interface{}) error
74+
FromData(input *commonpb.Payloads, valuePtrs ...interface{}) error
7575
}
7676

77-
// defaultDataConverter uses JSON.
78-
defaultDataConverter struct{}
77+
// PayloadConverter converts single value to/from payload.
78+
PayloadConverter interface {
79+
// ToData single value to payload.
80+
ToData(value interface{}) (*commonpb.Payload, error)
81+
// FromData single value from payload.
82+
FromData(input *commonpb.Payload, valuePtr interface{}) error
83+
}
84+
85+
defaultPayloadConverter struct{}
7986

80-
// NameValuePair represent named value.
81-
NameValuePair struct {
82-
Name string
83-
Value interface{}
87+
defaultDataConverter struct {
88+
payloadConverter PayloadConverter
8489
}
8590
)
8691

8792
var (
88-
// DefaultDataConverter is default data converter used by Temporal worker
89-
DefaultDataConverter = &defaultDataConverter{}
93+
// DefaultPayloadConverter is default single value serializer.
94+
DefaultPayloadConverter = &defaultPayloadConverter{}
95+
96+
// DefaultDataConverter is default data converter used by Temporal worker.
97+
DefaultDataConverter = &defaultDataConverter{
98+
payloadConverter: DefaultPayloadConverter,
99+
}
90100

91101
// ErrMetadataIsNotSet is returned when metadata is not set.
92102
ErrMetadataIsNotSet = errors.New("metadata is not set")
@@ -102,7 +112,7 @@ var (
102112
ErrUnableToSetBytes = errors.New("unable to set []byte value")
103113
)
104114

105-
// getDefaultDataConverter return default data converter used by Temporal worker
115+
// getDefaultDataConverter return default data converter used by Temporal worker.
106116
func getDefaultDataConverter() DataConverter {
107117
return DefaultDataConverter
108118
}
@@ -114,34 +124,11 @@ func (dc *defaultDataConverter) ToData(values ...interface{}) (*commonpb.Payload
114124

115125
result := &commonpb.Payloads{}
116126
for i, value := range values {
117-
nvp, ok := value.(NameValuePair)
118-
if !ok {
119-
nvp.Name = fmt.Sprintf("values[%d]", i)
120-
nvp.Value = value
127+
payload, err := dc.payloadConverter.ToData(value)
128+
if err != nil {
129+
return nil, fmt.Errorf("values[%d]: %w", i, err)
121130
}
122131

123-
var payload *commonpb.Payload
124-
if bytes, isByteSlice := nvp.Value.([]byte); isByteSlice {
125-
payload = &commonpb.Payload{
126-
Metadata: map[string][]byte{
127-
metadataEncoding: []byte(metadataEncodingRaw),
128-
metadataName: []byte(nvp.Name),
129-
},
130-
Data: bytes,
131-
}
132-
} else {
133-
data, err := json.Marshal(nvp.Value)
134-
if err != nil {
135-
return nil, fmt.Errorf("%s: %w: %v", nvp.Name, ErrUnableToEncodeJSON, err)
136-
}
137-
payload = &commonpb.Payload{
138-
Metadata: map[string][]byte{
139-
metadataEncoding: []byte(metadataEncodingJSON),
140-
metadataName: []byte(nvp.Name),
141-
},
142-
Data: data,
143-
}
144-
}
145132
result.Payloads = append(result.Payloads, payload)
146133
}
147134

@@ -158,40 +145,71 @@ func (dc *defaultDataConverter) FromData(payloads *commonpb.Payloads, valuePtrs
158145
break
159146
}
160147

161-
metadata := payload.GetMetadata()
162-
if metadata == nil {
163-
return fmt.Errorf("payload item %d: %w", i, ErrMetadataIsNotSet)
148+
err := dc.payloadConverter.FromData(payload, valuePtrs[i])
149+
if err != nil {
150+
return fmt.Errorf("payload item %d: %w", i, err)
164151
}
152+
}
165153

166-
var name string
167-
if n, ok := metadata[metadataName]; ok {
168-
name = string(n)
169-
} else {
170-
name = fmt.Sprintf("values[%d]", i)
171-
}
154+
return nil
155+
}
172156

173-
var encoding string
174-
if e, ok := metadata[metadataEncoding]; ok {
175-
encoding = string(e)
176-
} else {
177-
return fmt.Errorf("%s: %w", name, ErrEncodingIsNotSet)
157+
func (vs *defaultPayloadConverter) ToData(value interface{}) (*commonpb.Payload, error) {
158+
var payload *commonpb.Payload
159+
if bytes, isByteSlice := value.([]byte); isByteSlice {
160+
payload = &commonpb.Payload{
161+
Metadata: map[string][]byte{
162+
metadataEncoding: []byte(metadataEncodingRaw),
163+
},
164+
Data: bytes,
165+
}
166+
} else {
167+
data, err := json.Marshal(value)
168+
if err != nil {
169+
return nil, fmt.Errorf("%w: %v", ErrUnableToEncodeJSON, err)
170+
}
171+
payload = &commonpb.Payload{
172+
Metadata: map[string][]byte{
173+
metadataEncoding: []byte(metadataEncodingJSON),
174+
},
175+
Data: data,
178176
}
177+
}
178+
179+
return payload, nil
180+
}
181+
182+
func (vs *defaultPayloadConverter) FromData(payload *commonpb.Payload, valuePtr interface{}) error {
183+
if payload == nil {
184+
return nil
185+
}
179186

180-
switch encoding {
181-
case metadataEncodingRaw:
182-
valueBytes := reflect.ValueOf(valuePtrs[i]).Elem()
183-
if !valueBytes.CanSet() {
184-
return fmt.Errorf("%s: %w", name, ErrUnableToSetBytes)
185-
}
186-
valueBytes.SetBytes(payload.GetData())
187-
case metadataEncodingJSON:
188-
err := json.Unmarshal(payload.GetData(), valuePtrs[i])
189-
if err != nil {
190-
return fmt.Errorf("%s: %w: %v", name, ErrUnableToDecodeJSON, err)
191-
}
192-
default:
193-
return fmt.Errorf("%s, encoding %s: %w", name, encoding, ErrEncodingIsNotSupported)
187+
metadata := payload.GetMetadata()
188+
if metadata == nil {
189+
return ErrMetadataIsNotSet
190+
}
191+
192+
var encoding string
193+
if e, ok := metadata[metadataEncoding]; ok {
194+
encoding = string(e)
195+
} else {
196+
return ErrEncodingIsNotSet
197+
}
198+
199+
switch encoding {
200+
case metadataEncodingRaw:
201+
valueBytes := reflect.ValueOf(valuePtr).Elem()
202+
if !valueBytes.CanSet() {
203+
return ErrUnableToSetBytes
204+
}
205+
valueBytes.SetBytes(payload.GetData())
206+
case metadataEncodingJSON:
207+
err := json.Unmarshal(payload.GetData(), valuePtr)
208+
if err != nil {
209+
return fmt.Errorf("%w: %v", ErrUnableToDecodeJSON, err)
194210
}
211+
default:
212+
return fmt.Errorf("encoding %s: %w", encoding, ErrEncodingIsNotSupported)
195213
}
196214

197215
return nil

‎internal/error_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -438,10 +438,10 @@ func Test_ContinueAsNewError(t *testing.T) {
438438
return NewContinueAsNewError(ctx, continueAsNewWfName, a1, a2)
439439
}
440440

441-
headerValue, err := DefaultDataConverter.ToData("test-data")
441+
headerValue, err := DefaultPayloadConverter.ToData("test-data")
442442
assert.NoError(t, err)
443443
header := &commonpb.Header{
444-
Fields: map[string]*commonpb.Payloads{"test": headerValue},
444+
Fields: map[string]*commonpb.Payload{"test": headerValue},
445445
}
446446

447447
s := &WorkflowTestSuite{

‎internal/headers.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -32,12 +32,12 @@ import (
3232

3333
// HeaderWriter is an interface to write information to temporal headers
3434
type HeaderWriter interface {
35-
Set(string, *commonpb.Payloads)
35+
Set(string, *commonpb.Payload)
3636
}
3737

3838
// HeaderReader is an interface to read information from temporal headers
3939
type HeaderReader interface {
40-
ForEachKey(handler func(string, *commonpb.Payloads) error) error
40+
ForEachKey(handler func(string, *commonpb.Payload) error) error
4141
}
4242

4343
// ContextPropagator is an interface that determines what information from
@@ -62,7 +62,7 @@ type headerReader struct {
6262
header *commonpb.Header
6363
}
6464

65-
func (hr *headerReader) ForEachKey(handler func(string, *commonpb.Payloads) error) error {
65+
func (hr *headerReader) ForEachKey(handler func(string, *commonpb.Payload) error) error {
6666
if hr.header == nil {
6767
return nil
6868
}
@@ -83,7 +83,7 @@ type headerWriter struct {
8383
header *commonpb.Header
8484
}
8585

86-
func (hw *headerWriter) Set(key string, value *commonpb.Payloads) {
86+
func (hw *headerWriter) Set(key string, value *commonpb.Payload) {
8787
if hw.header == nil {
8888
return
8989
}
@@ -93,7 +93,7 @@ func (hw *headerWriter) Set(key string, value *commonpb.Payloads) {
9393
// NewHeaderWriter returns a header writer interface
9494
func NewHeaderWriter(header *commonpb.Header) HeaderWriter {
9595
if header != nil && header.Fields == nil {
96-
header.Fields = make(map[string]*commonpb.Payloads)
96+
header.Fields = make(map[string]*commonpb.Payload)
9797
}
9898
return &headerWriter{header}
9999
}

‎internal/headers_test.go‎

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -37,48 +37,48 @@ func TestHeaderWriter(t *testing.T) {
3737
name string
3838
initial *commonpb.Header
3939
expected *commonpb.Header
40-
vals map[string]*commonpb.Payloads
40+
vals map[string]*commonpb.Payload
4141
}{
4242
{
4343
"no values",
4444
&commonpb.Header{
45-
Fields: map[string]*commonpb.Payloads{},
45+
Fields: map[string]*commonpb.Payload{},
4646
},
4747
&commonpb.Header{
48-
Fields: map[string]*commonpb.Payloads{},
48+
Fields: map[string]*commonpb.Payload{},
4949
},
50-
map[string]*commonpb.Payloads{},
50+
map[string]*commonpb.Payload{},
5151
},
5252
{
5353
"add values",
5454
&commonpb.Header{
55-
Fields: map[string]*commonpb.Payloads{},
55+
Fields: map[string]*commonpb.Payload{},
5656
},
5757
&commonpb.Header{
58-
Fields: map[string]*commonpb.Payloads{
58+
Fields: map[string]*commonpb.Payload{
5959
"key1": encodeString(t, "val1"),
6060
"key2": encodeString(t, "val2"),
6161
},
6262
},
63-
map[string]*commonpb.Payloads{
63+
map[string]*commonpb.Payload{
6464
"key1": encodeString(t, "val1"),
6565
"key2": encodeString(t, "val2"),
6666
},
6767
},
6868
{
6969
"overwrite values",
7070
&commonpb.Header{
71-
Fields: map[string]*commonpb.Payloads{
71+
Fields: map[string]*commonpb.Payload{
7272
"key1": encodeString(t, "unexpected"),
7373
},
7474
},
7575
&commonpb.Header{
76-
Fields: map[string]*commonpb.Payloads{
76+
Fields: map[string]*commonpb.Payload{
7777
"key1": encodeString(t, "val1"),
7878
"key2": encodeString(t, "val2"),
7979
},
8080
},
81-
map[string]*commonpb.Payloads{
81+
map[string]*commonpb.Payload{
8282
"key1": encodeString(t, "val1"),
8383
"key2": encodeString(t, "val2"),
8484
},
@@ -98,8 +98,8 @@ func TestHeaderWriter(t *testing.T) {
9898
}
9999
}
100100

101-
func encodeString(t *testing.T, s string) *commonpb.Payloads {
102-
p, err := DefaultDataConverter.ToData(s)
101+
func encodeString(t *testing.T, s string) *commonpb.Payload {
102+
p, err := DefaultPayloadConverter.ToData(s)
103103
assert.NoError(t, err)
104104
return p
105105
}
@@ -115,7 +115,7 @@ func TestHeaderReader(t *testing.T) {
115115
{
116116
"valid values",
117117
&commonpb.Header{
118-
Fields: map[string]*commonpb.Payloads{
118+
Fields: map[string]*commonpb.Payload{
119119
"key1": encodeString(t, "val1"),
120120
"key2": encodeString(t, "val2"),
121121
},
@@ -126,7 +126,7 @@ func TestHeaderReader(t *testing.T) {
126126
{
127127
"invalid values",
128128
&commonpb.Header{
129-
Fields: map[string]*commonpb.Payloads{
129+
Fields: map[string]*commonpb.Payload{
130130
"key1": encodeString(t, "val1"),
131131
"key2": encodeString(t, "val2"),
132132
},
@@ -141,7 +141,7 @@ func TestHeaderReader(t *testing.T) {
141141
t.Run(test.name, func(t *testing.T) {
142142
t.Parallel()
143143
reader := NewHeaderReader(test.header)
144-
err := reader.ForEachKey(func(key string, _ *commonpb.Payloads) error {
144+
err := reader.ForEachKey(func(key string, _ *commonpb.Payload) error {
145145
if _, ok := test.keys[key]; !ok {
146146
return assert.AnError
147147
}

‎internal/internal_event_handlers.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -326,7 +326,7 @@ func mergeSearchAttributes(current, upsert *commonpb.SearchAttributes) *commonpb
326326
return nil
327327
}
328328
current = &commonpb.SearchAttributes{
329-
IndexedFields: make(map[string]*commonpb.Payloads),
329+
IndexedFields: make(map[string]*commonpb.Payload),
330330
}
331331
}
332332

0 commit comments

Comments
 (0)