018ef47c-7ef1-4ce1-b4c9-3fa25bfb5bee

Committed 19 Apr 2024 03:53AM UTC coverage: 67.594% (+0.02%) from 67.579%

Build # 018ef47c-7ef1-4ce1-b4c9-3fa25bfb5bee

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Add document explaining the schema of cassandra executions table (#5921)

Run Details

98984 of 146438 relevant lines covered (67.59%)

2380.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.13

/common/asyncworkflow/queue/consumer/default_consumer.go

// The MIT License (MIT)

// Copyright (c) 2017-2020 Uber Technologies Inc.

// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.

package consumer

import (
        "context"
        "fmt"
        "sort"
        "sync"
        "time"

        "go.uber.org/yarpc"

        "github.com/uber/cadence/.gen/go/shared"
        "github.com/uber/cadence/.gen/go/sqlblobs"
        "github.com/uber/cadence/client/frontend"
        "github.com/uber/cadence/common"
        "github.com/uber/cadence/common/backoff"
        "github.com/uber/cadence/common/codec"
        "github.com/uber/cadence/common/log"
        "github.com/uber/cadence/common/log/tag"
        "github.com/uber/cadence/common/messaging"
        "github.com/uber/cadence/common/metrics"
        "github.com/uber/cadence/common/types"
        "github.com/uber/cadence/common/types/mapper/thrift"
)

const (
        defaultShutdownTimeout = 5 * time.Second
        defaultStartWFTimeout  = 3 * time.Second
        defaultConcurrency     = 100
)

type DefaultConsumer struct {
        queueID         string
        innerConsumer   messaging.Consumer
        logger          log.Logger
        scope           metrics.Scope
        frontendClient  frontend.Client
        ctx             context.Context
        cancelFn        context.CancelFunc
        wg              sync.WaitGroup
        shutdownTimeout time.Duration
        startWFTimeout  time.Duration
        msgDecoder      codec.BinaryEncoder
        concurrency     int
}

type Option func(*DefaultConsumer)

func WithConcurrency(concurrency int) Option {
        return func(c *DefaultConsumer) {
                c.concurrency = concurrency
        }
}

func New(
        queueID string,
        innerConsumer messaging.Consumer,
        logger log.Logger,
        metricsClient metrics.Client,
        frontendClient frontend.Client,
        options ...Option,
) *DefaultConsumer {
        ctx, cancelFn := context.WithCancel(context.Background())
        c := &DefaultConsumer{
                queueID:         queueID,
                innerConsumer:   innerConsumer,
                logger:          logger.WithTags(tag.AsyncWFQueueID(queueID)),
                scope:           metricsClient.Scope(metrics.AsyncWorkflowConsumerScope),
                frontendClient:  frontendClient,
                ctx:             ctx,
                cancelFn:        cancelFn,
                shutdownTimeout: defaultShutdownTimeout,
                startWFTimeout:  defaultStartWFTimeout,
                msgDecoder:      codec.NewThriftRWEncoder(),
                concurrency:     defaultConcurrency,
        }

        for _, opt := range options {
                opt(c)
        }

        return c
}

func (c *DefaultConsumer) Start() error {
        if err := c.innerConsumer.Start(); err != nil {
                return err
        }

        for i := 0; i < c.concurrency; i++ {
                c.wg.Add(1)
                go c.runProcessLoop()
                c.logger.Info("Started process loop", tag.Counter(i))
        }
        c.logger.Info("Started consumer", tag.Dynamic("concurrency", c.concurrency))
        return nil
}

func (c *DefaultConsumer) Stop() {
        c.logger.Info("Stopping consumer")
        c.cancelFn()
        c.wg.Wait()
        if !common.AwaitWaitGroup(&c.wg, c.shutdownTimeout) {
                c.logger.Warn("Consumer timed out on shutdown", tag.Dynamic("timeout", c.shutdownTimeout))
                return
        }

        c.innerConsumer.Stop()
        c.logger.Info("Stopped consumer")
}

func (c *DefaultConsumer) runProcessLoop() {
        defer c.wg.Done()

        for {
                select {
                case msg, ok := <-c.innerConsumer.Messages():
                        if !ok {
                                c.logger.Info("Consumer channel closed")
                                return
                        }

                        c.processMessage(msg)
                case <-c.ctx.Done():
                        c.logger.Info("Consumer context done so terminating loop")
                        return
                }
        }
}

func (c *DefaultConsumer) processMessage(msg messaging.Message) {
        logger := c.logger.WithTags(tag.Dynamic("partition", msg.Partition()), tag.Dynamic("offset", msg.Offset()))
        logger.Debug("Received message")

        sw := c.scope.StartTimer(metrics.AsyncWorkflowProcessMsgLatency)
        defer sw.Stop()

        var request sqlblobs.AsyncRequestMessage
        if err := c.msgDecoder.Decode(msg.Value(), &request); err != nil {
                logger.Error("Failed to decode message", tag.Error(err))
                c.scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)
                if err := msg.Nack(); err != nil {
                        logger.Error("Failed to nack message", tag.Error(err))
                }
                return
        }

        if err := c.processRequest(logger, &request); err != nil {
                logger.Error("Failed to process message", tag.Error(err))
                if err := msg.Nack(); err != nil {
                        logger.Error("Failed to nack message", tag.Error(err))
                }
                return
        }

        if err := msg.Ack(); err != nil {
                logger.Error("Failed to ack message", tag.Error(err))
        }
        logger.Debug("Processed message successfully")
}

func (c *DefaultConsumer) processRequest(logger log.Logger, request *sqlblobs.AsyncRequestMessage) error {
        scope := c.scope.Tagged(metrics.AsyncWFRequestTypeTag(request.GetType().String()))
        switch request.GetType() {
        case sqlblobs.AsyncRequestTypeStartWorkflowExecutionAsyncRequest:
                startWFReq, err := c.decodeStartWorkflowRequest(request.GetPayload(), request.GetEncoding())
                if err != nil {
                        scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)
                        return err
                }

                yarpcCallOpts := getYARPCOptions(request.GetHeader())
                scope := scope.Tagged(metrics.DomainTag(startWFReq.GetDomain()))

                var resp *types.StartWorkflowExecutionResponse
                op := func() error {
                        ctx, cancel := context.WithTimeout(c.ctx, c.startWFTimeout)
                        defer cancel()
                        resp, err = c.frontendClient.StartWorkflowExecution(ctx, startWFReq, yarpcCallOpts...)
                        return err
                }

                if err := callFrontendWithRetries(c.ctx, op); err != nil {
                        scope.IncCounter(metrics.AsyncWorkflowFailureByFrontendCount)
                        return fmt.Errorf("start workflow execution failed after all attempts: %w", err)
                }

                scope.IncCounter(metrics.AsyncWorkflowSuccessCount)
                logger.Info("StartWorkflowExecution succeeded", tag.WorkflowID(startWFReq.GetWorkflowID()), tag.WorkflowRunID(resp.GetRunID()))
        case sqlblobs.AsyncRequestTypeSignalWithStartWorkflowExecutionAsyncRequest:
                startWFReq, err := c.decodeSignalWithStartWorkflowRequest(request.GetPayload(), request.GetEncoding())
                if err != nil {
                        c.scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)
                        return err
                }

                yarpcCallOpts := getYARPCOptions(request.GetHeader())
                scope := c.scope.Tagged(metrics.DomainTag(startWFReq.GetDomain()))
                var resp *types.StartWorkflowExecutionResponse
                op := func() error {
                        ctx, cancel := context.WithTimeout(c.ctx, c.startWFTimeout)
                        defer cancel()
                        resp, err = c.frontendClient.SignalWithStartWorkflowExecution(ctx, startWFReq, yarpcCallOpts...)
                        return err
                }

                if err := callFrontendWithRetries(c.ctx, op); err != nil {
                        scope.IncCounter(metrics.AsyncWorkflowFailureByFrontendCount)
                        return fmt.Errorf("signal with start workflow execution failed after all attempts: %w", err)
                }

                scope.IncCounter(metrics.AsyncWorkflowSuccessCount)
                logger.Info("SignalWithStartWorkflowExecution succeeded", tag.WorkflowID(startWFReq.GetWorkflowID()), tag.WorkflowRunID(resp.GetRunID()))
        default:
                c.scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)
                return &UnsupportedRequestType{Type: request.GetType()}
        }

        return nil
}

func callFrontendWithRetries(ctx context.Context, op func() error) error {
        throttleRetry := backoff.NewThrottleRetry(
                backoff.WithRetryPolicy(common.CreateFrontendServiceRetryPolicy()),
                backoff.WithRetryableError(common.IsServiceTransientError),
        )

        return throttleRetry.Do(ctx, op)
}

func getYARPCOptions(header *shared.Header) []yarpc.CallOption {
        if header == nil || header.GetFields() == nil {
                return nil
        }

        // sort the header fields to make the tests deterministic
        fields := header.GetFields()
        sortedKeys := make([]string, 0, len(fields))
        for k := range fields {
                sortedKeys = append(sortedKeys, k)
        }
        sort.Strings(sortedKeys)

        var opts []yarpc.CallOption
        for _, k := range sortedKeys {
                opts = append(opts, yarpc.WithHeader(k, string(fields[k])))
        }
        return opts
}

func (c *DefaultConsumer) decodeStartWorkflowRequest(payload []byte, encoding string) (*types.StartWorkflowExecutionRequest, error) {
        if encoding != string(common.EncodingTypeThriftRW) {
                return nil, &UnsupportedEncoding{EncodingType: encoding}
        }

        var thriftObj shared.StartWorkflowExecutionAsyncRequest
        if err := c.msgDecoder.Decode(payload, &thriftObj); err != nil {
                return nil, err
        }

        startRequest := thrift.ToStartWorkflowExecutionAsyncRequest(&thriftObj)
        return startRequest.StartWorkflowExecutionRequest, nil
}

func (c *DefaultConsumer) decodeSignalWithStartWorkflowRequest(payload []byte, encoding string) (*types.SignalWithStartWorkflowExecutionRequest, error) {
        if encoding != string(common.EncodingTypeThriftRW) {
                return nil, &UnsupportedEncoding{EncodingType: encoding}
        }

        var thriftObj shared.SignalWithStartWorkflowExecutionAsyncRequest
        if err := c.msgDecoder.Decode(payload, &thriftObj); err != nil {
                return nil, err
        }

        signalWithStartRequest := thrift.ToSignalWithStartWorkflowExecutionAsyncRequest(&thriftObj)
        return signalWithStartRequest.SignalWithStartWorkflowExecutionRequest, nil
}

1	// The MIT License (MIT)
2
3	// Copyright (c) 2017-2020 Uber Technologies Inc.
4
5	// Permission is hereby granted, free of charge, to any person obtaining a copy
6	// of this software and associated documentation files (the "Software"), to deal
7	// in the Software without restriction, including without limitation the rights
8	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9	// copies of the Software, and to permit persons to whom the Software is
10	// furnished to do so, subject to the following conditions:
11	//
12	// The above copyright notice and this permission notice shall be included in all
13	// copies or substantial portions of the Software.
14	//
15	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21	// SOFTWARE.
22
23	package consumer
24
25	import (
26	"context"
27	"fmt"
28	"sort"
29	"sync"
30	"time"
31
32	"go.uber.org/yarpc"
33
34	"github.com/uber/cadence/.gen/go/shared"
35	"github.com/uber/cadence/.gen/go/sqlblobs"
36	"github.com/uber/cadence/client/frontend"
37	"github.com/uber/cadence/common"
38	"github.com/uber/cadence/common/backoff"
39	"github.com/uber/cadence/common/codec"
40	"github.com/uber/cadence/common/log"
41	"github.com/uber/cadence/common/log/tag"
42	"github.com/uber/cadence/common/messaging"
43	"github.com/uber/cadence/common/metrics"
44	"github.com/uber/cadence/common/types"
45	"github.com/uber/cadence/common/types/mapper/thrift"
46	)
47
48	const (
49	defaultShutdownTimeout = 5 * time.Second
50	defaultStartWFTimeout = 3 * time.Second
51	defaultConcurrency = 100
52	)
53
54	type DefaultConsumer struct {
55	queueID string
56	innerConsumer messaging.Consumer
57	logger log.Logger
58	scope metrics.Scope
59	frontendClient frontend.Client
60	ctx context.Context
61	cancelFn context.CancelFunc
62	wg sync.WaitGroup
63	shutdownTimeout time.Duration
64	startWFTimeout time.Duration
65	msgDecoder codec.BinaryEncoder
66	concurrency int
67	}
68
69	type Option func(*DefaultConsumer)
70
71	func WithConcurrency(concurrency int) Option {	12✔
72	return func(c *DefaultConsumer) {	24✔
73	c.concurrency = concurrency	12✔
74	}	12✔
75	}
76
77	func New(
78	queueID string,
79	innerConsumer messaging.Consumer,
80	logger log.Logger,
81	metricsClient metrics.Client,
82	frontendClient frontend.Client,
83	options ...Option,
84	) *DefaultConsumer {	12✔
85	ctx, cancelFn := context.WithCancel(context.Background())	12✔
86	c := &DefaultConsumer{	12✔
87	queueID: queueID,	12✔
88	innerConsumer: innerConsumer,	12✔
89	logger: logger.WithTags(tag.AsyncWFQueueID(queueID)),	12✔
90	scope: metricsClient.Scope(metrics.AsyncWorkflowConsumerScope),	12✔
91	frontendClient: frontendClient,	12✔
92	ctx: ctx,	12✔
93	cancelFn: cancelFn,	12✔
94	shutdownTimeout: defaultShutdownTimeout,	12✔
95	startWFTimeout: defaultStartWFTimeout,	12✔
96	msgDecoder: codec.NewThriftRWEncoder(),	12✔
97	concurrency: defaultConcurrency,	12✔
98	}	12✔
99		12✔
100	for _, opt := range options {	24✔
101	opt(c)	12✔
102	}	12✔
103
104	return c	12✔
105	}
106
107	func (c *DefaultConsumer) Start() error {	12✔
108	if err := c.innerConsumer.Start(); err != nil {	13✔
109	return err	1✔
110	}	1✔
111
112	for i := 0; i < c.concurrency; i++ {	33✔
113	c.wg.Add(1)	22✔
114	go c.runProcessLoop()	22✔
115	c.logger.Info("Started process loop", tag.Counter(i))	22✔
116	}	22✔
117	c.logger.Info("Started consumer", tag.Dynamic("concurrency", c.concurrency))	11✔
118	return nil	11✔
119	}
120
121	func (c *DefaultConsumer) Stop() {	11✔
122	c.logger.Info("Stopping consumer")	11✔
123	c.cancelFn()	11✔
124	c.wg.Wait()	11✔
125	if !common.AwaitWaitGroup(&c.wg, c.shutdownTimeout) {	11✔
126	c.logger.Warn("Consumer timed out on shutdown", tag.Dynamic("timeout", c.shutdownTimeout))	×
127	return	×
128	}	×
129
130	c.innerConsumer.Stop()	11✔
131	c.logger.Info("Stopped consumer")	11✔
132	}
133
134	func (c *DefaultConsumer) runProcessLoop() {	22✔
135	defer c.wg.Done()	22✔
136		22✔
137	for {	56✔
138	select {	34✔
139	case msg, ok := <-c.innerConsumer.Messages():	12✔
140	if !ok {	12✔
141	c.logger.Info("Consumer channel closed")	×
142	return	×
143	}	×
144
145	c.processMessage(msg)	12✔
146	case <-c.ctx.Done():	22✔
147	c.logger.Info("Consumer context done so terminating loop")	22✔
148	return	22✔
149	}
150	}
151	}
152
153	func (c *DefaultConsumer) processMessage(msg messaging.Message) {	12✔
154	logger := c.logger.WithTags(tag.Dynamic("partition", msg.Partition()), tag.Dynamic("offset", msg.Offset()))	12✔
155	logger.Debug("Received message")	12✔
156		12✔
157	sw := c.scope.StartTimer(metrics.AsyncWorkflowProcessMsgLatency)	12✔
158	defer sw.Stop()	12✔
159		12✔
160	var request sqlblobs.AsyncRequestMessage	12✔
161	if err := c.msgDecoder.Decode(msg.Value(), &request); err != nil {	14✔
162	logger.Error("Failed to decode message", tag.Error(err))	2✔
163	c.scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)	2✔
164	if err := msg.Nack(); err != nil {	2✔
165	logger.Error("Failed to nack message", tag.Error(err))	×
166	}	×
167	return	2✔
168	}
169
170	if err := c.processRequest(logger, &request); err != nil {	17✔
171	logger.Error("Failed to process message", tag.Error(err))	7✔
172	if err := msg.Nack(); err != nil {	7✔
173	logger.Error("Failed to nack message", tag.Error(err))	×
174	}	×
175	return	7✔
176	}
177
178	if err := msg.Ack(); err != nil {	3✔
179	logger.Error("Failed to ack message", tag.Error(err))	×
180	}	×
181	logger.Debug("Processed message successfully")	3✔
182	}
183
184	func (c DefaultConsumer) processRequest(logger log.Logger, request sqlblobs.AsyncRequestMessage) error {	10✔
185	scope := c.scope.Tagged(metrics.AsyncWFRequestTypeTag(request.GetType().String()))	10✔
186	switch request.GetType() {	10✔
187	case sqlblobs.AsyncRequestTypeStartWorkflowExecutionAsyncRequest:	5✔
188	startWFReq, err := c.decodeStartWorkflowRequest(request.GetPayload(), request.GetEncoding())	5✔
189	if err != nil {	7✔
190	scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)	2✔
191	return err	2✔
192	}	2✔
193
194	yarpcCallOpts := getYARPCOptions(request.GetHeader())	3✔
195	scope := scope.Tagged(metrics.DomainTag(startWFReq.GetDomain()))	3✔
196		3✔
197	var resp *types.StartWorkflowExecutionResponse	3✔
198	op := func() error {	6✔
199	ctx, cancel := context.WithTimeout(c.ctx, c.startWFTimeout)	3✔
200	defer cancel()	3✔
201	resp, err = c.frontendClient.StartWorkflowExecution(ctx, startWFReq, yarpcCallOpts...)	3✔
202	return err	3✔
203	}	3✔
204
205	if err := callFrontendWithRetries(c.ctx, op); err != nil {	4✔
206	scope.IncCounter(metrics.AsyncWorkflowFailureByFrontendCount)	1✔
207	return fmt.Errorf("start workflow execution failed after all attempts: %w", err)	1✔
208	}	1✔
209
210	scope.IncCounter(metrics.AsyncWorkflowSuccessCount)	2✔
211	logger.Info("StartWorkflowExecution succeeded", tag.WorkflowID(startWFReq.GetWorkflowID()), tag.WorkflowRunID(resp.GetRunID()))	2✔
212	case sqlblobs.AsyncRequestTypeSignalWithStartWorkflowExecutionAsyncRequest:	4✔
213	startWFReq, err := c.decodeSignalWithStartWorkflowRequest(request.GetPayload(), request.GetEncoding())	4✔
214	if err != nil {	6✔
215	c.scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)	2✔
216	return err	2✔
217	}	2✔
218
219	yarpcCallOpts := getYARPCOptions(request.GetHeader())	2✔
220	scope := c.scope.Tagged(metrics.DomainTag(startWFReq.GetDomain()))	2✔
221	var resp *types.StartWorkflowExecutionResponse	2✔
222	op := func() error {	4✔
223	ctx, cancel := context.WithTimeout(c.ctx, c.startWFTimeout)	2✔
224	defer cancel()	2✔
225	resp, err = c.frontendClient.SignalWithStartWorkflowExecution(ctx, startWFReq, yarpcCallOpts...)	2✔
226	return err	2✔
227	}	2✔
228
229	if err := callFrontendWithRetries(c.ctx, op); err != nil {	3✔
230	scope.IncCounter(metrics.AsyncWorkflowFailureByFrontendCount)	1✔
231	return fmt.Errorf("signal with start workflow execution failed after all attempts: %w", err)	1✔
232	}	1✔
233
234	scope.IncCounter(metrics.AsyncWorkflowSuccessCount)	1✔
235	logger.Info("SignalWithStartWorkflowExecution succeeded", tag.WorkflowID(startWFReq.GetWorkflowID()), tag.WorkflowRunID(resp.GetRunID()))	1✔
236	default:	1✔
237	c.scope.IncCounter(metrics.AsyncWorkflowFailureCorruptMsgCount)	1✔
238	return &UnsupportedRequestType{Type: request.GetType()}	1✔
239	}
240
241	return nil	3✔
242	}
243
244	func callFrontendWithRetries(ctx context.Context, op func() error) error {	5✔
245	throttleRetry := backoff.NewThrottleRetry(	5✔
246	backoff.WithRetryPolicy(common.CreateFrontendServiceRetryPolicy()),	5✔
247	backoff.WithRetryableError(common.IsServiceTransientError),	5✔
248	)	5✔
249		5✔
250	return throttleRetry.Do(ctx, op)	5✔
251	}	5✔
252
253	func getYARPCOptions(header *shared.Header) []yarpc.CallOption {	17✔
254	if header == nil \|\| header.GetFields() == nil {	17✔
255	return nil	×
256	}	×
257
258	// sort the header fields to make the tests deterministic
259	fields := header.GetFields()	17✔
260	sortedKeys := make([]string, 0, len(fields))	17✔
261	for k := range fields {	51✔
262	sortedKeys = append(sortedKeys, k)	34✔
263	}	34✔
264	sort.Strings(sortedKeys)	17✔
265		17✔
266	var opts []yarpc.CallOption	17✔
267	for _, k := range sortedKeys {	51✔
268	opts = append(opts, yarpc.WithHeader(k, string(fields[k])))	34✔
269	}	34✔
270	return opts	17✔
271	}
272
273	func (c DefaultConsumer) decodeStartWorkflowRequest(payload []byte, encoding string) (types.StartWorkflowExecutionRequest, error) {	5✔
274	if encoding != string(common.EncodingTypeThriftRW) {	6✔
275	return nil, &UnsupportedEncoding{EncodingType: encoding}	1✔
276	}	1✔
277
278	var thriftObj shared.StartWorkflowExecutionAsyncRequest	4✔
279	if err := c.msgDecoder.Decode(payload, &thriftObj); err != nil {	5✔
280	return nil, err	1✔
281	}	1✔
282
283	startRequest := thrift.ToStartWorkflowExecutionAsyncRequest(&thriftObj)	3✔
284	return startRequest.StartWorkflowExecutionRequest, nil	3✔
285	}
286
287	func (c DefaultConsumer) decodeSignalWithStartWorkflowRequest(payload []byte, encoding string) (types.SignalWithStartWorkflowExecutionRequest, error) {	4✔
288	if encoding != string(common.EncodingTypeThriftRW) {	5✔
289	return nil, &UnsupportedEncoding{EncodingType: encoding}	1✔
290	}	1✔
291
292	var thriftObj shared.SignalWithStartWorkflowExecutionAsyncRequest	3✔
293	if err := c.msgDecoder.Decode(payload, &thriftObj); err != nil {	4✔
294	return nil, err	1✔
295	}	1✔
296
297	signalWithStartRequest := thrift.ToSignalWithStartWorkflowExecutionAsyncRequest(&thriftObj)	2✔
298	return signalWithStartRequest.SignalWithStartWorkflowExecutionRequest, nil	2✔
299	}

uber / cadence / 018ef47c-7ef1-4ce1-b4c9-3fa25bfb5bee

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous