func (wc *workflowClient) CompleteActivity(ctx context.Context, taskToken []byte, result interface{}, err error) error {
	if taskToken == nil {
		return errors.New("invalid task token provided")
	}
	var data []byte
	if result != nil {
		var err0 error
		data, err0 = encodeArg(wc.dataConverter, result)
		if err0 != nil {
			return err0
		}
	}
	request := convertActivityResultToRespondRequest(wc.identity, taskToken, data, err, wc.dataConverter)
	return reportActivityComplete(ctx, wc.workflowService, request, wc.metricsScope, wc.featureFlags)
}
CompleteActivity reports activity completed. activity Execute method can return activity.ErrResultPending to
indicate the activity is not completed when it's Execute method returns. In that case, this CompleteActivity() method
should be called when that activity is completed with the actual result and error. If err is nil, activity task
completed event will be reported; if err is CanceledError, activity task cancelled event will be reported; otherwise,
activity task failed event will be reported.
package internal
import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"time"
	"go.uber.org/cadence/internal/common/serializer"
	"github.com/opentracing/opentracing-go"
	"github.com/pborman/uuid"
	"github.com/uber-go/tally"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	s "go.uber.org/cadence/.gen/go/shared"
	"go.uber.org/cadence/internal/common"
	"go.uber.org/cadence/internal/common/backoff"
	"go.uber.org/cadence/internal/common/metrics"
)
var _ Client = (*workflowClient)(nil)
var _ DomainClient = (*domainClient)(nil)
const (
	defaultDecisionTaskTimeoutInSecs = 10
	defaultGetHistoryTimeoutInSecs   = 25
)
var (
	maxListArchivedWorkflowTimeout = time.Minute * 3
)
type (
	workflowClient struct {
		workflowService    workflowserviceclient.Interface
		domain             string
		registry           *registry
		metricsScope       *metrics.TaggedScope
		identity           string
		dataConverter      DataConverter
		contextPropagators []ContextPropagator
		tracer             opentracing.Tracer
		featureFlags       FeatureFlags
	}
	domainClient struct {
		workflowService workflowserviceclient.Interface
		metricsScope    tally.Scope
		identity        string
		featureFlags    FeatureFlags
	}
	WorkflowRun interface {
		GetID() string
		GetRunID() string
		Get(ctx context.Context, valuePtr interface{}) error
	}
	workflowRunImpl struct {
		workflowFn    interface{}
		workflowID    string
		firstRunID    string
		currentRunID  string
		iterFn        func(ctx context.Context, runID string) HistoryEventIterator
		dataConverter DataConverter
		registry      *registry
	}
	HistoryEventIterator interface {
		HasNext() bool
		Next() (*s.HistoryEvent, error)
	}
	historyEventIteratorImpl struct {
		initialized bool
		nextEventIndex int
		events         []*s.HistoryEvent
		nexttoken []byte
		err error
		paginate func(nexttoken []byte) (*s.GetWorkflowExecutionHistoryResponse, error)
	}
)
func (wc *workflowClient) StartWorkflow(
	ctx context.Context,
	options StartWorkflowOptions,
	workflowFunc interface{},
	args ...interface{},
) (*WorkflowExecution, error) {
	workflowID := options.ID
	if len(workflowID) == 0 {
		workflowID = uuid.NewRandom().String()
	}
	if options.TaskList == "" {
		return nil, errors.New("missing TaskList")
	}
	executionTimeout := common.Int32Ceil(options.ExecutionStartToCloseTimeout.Seconds())
	if executionTimeout <= 0 {
		return nil, errors.New("missing or invalid ExecutionStartToCloseTimeout")
	}
	decisionTaskTimeout := common.Int32Ceil(options.DecisionTaskStartToCloseTimeout.Seconds())
	if decisionTaskTimeout < 0 {
		return nil, errors.New("negative DecisionTaskStartToCloseTimeout provided")
	}
	if decisionTaskTimeout == 0 {
		decisionTaskTimeout = defaultDecisionTaskTimeoutInSecs
	}
	workflowType, input, err := getValidatedWorkflowFunction(workflowFunc, args, wc.dataConverter, wc.registry)
	if err != nil {
		return nil, err
	}
	memo, err := getWorkflowMemo(options.Memo, wc.dataConverter)
	if err != nil {
		return nil, err
	}
	searchAttr, err := serializeSearchAttributes(options.SearchAttributes)
	if err != nil {
		return nil, err
	}
	delayStartSeconds := common.Int32Ceil(options.DelayStart.Seconds())
	if delayStartSeconds < 0 {
		return nil, errors.New("Invalid DelayStart option")
	}
	ctx, span := createOpenTracingWorkflowSpan(ctx, wc.tracer, time.Now(), fmt.Sprintf("StartWorkflow-%s", workflowType.Name), workflowID)
	span.Finish()
	header := wc.getWorkflowHeader(ctx)
	startRequest := &s.StartWorkflowExecutionRequest{
		Domain:                              common.StringPtr(wc.domain),
		RequestId:                           common.StringPtr(uuid.New()),
		WorkflowId:                          common.StringPtr(workflowID),
		WorkflowType:                        workflowTypePtr(*workflowType),
		TaskList:                            common.TaskListPtr(s.TaskList{Name: common.StringPtr(options.TaskList)}),
		Input:                               input,
		ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(executionTimeout),
		TaskStartToCloseTimeoutSeconds:      common.Int32Ptr(decisionTaskTimeout),
		Identity:                            common.StringPtr(wc.identity),
		WorkflowIdReusePolicy:               options.WorkflowIDReusePolicy.toThriftPtr(),
		RetryPolicy:                         convertRetryPolicy(options.RetryPolicy),
		CronSchedule:                        common.StringPtr(options.CronSchedule),
		Memo:                                memo,
		SearchAttributes:                    searchAttr,
		Header:                              header,
		DelayStartSeconds:                   common.Int32Ptr(delayStartSeconds),
	}
	var response *s.StartWorkflowExecutionResponse
	err = backoff.Retry(ctx,
		func() error {
			tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags)
			defer cancel()
			var err1 error
			response, err1 = wc.workflowService.StartWorkflowExecution(tchCtx, startRequest, opt...)
			return err1
		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
	if err != nil {
		return nil, err
	}
	if wc.metricsScope != nil {
		scope := wc.metricsScope.GetTaggedScope(tagTaskList, options.TaskList, tagWorkflowType, workflowType.Name)
		scope.Counter(metrics.WorkflowStartCounter).Inc(1)
	}
	executionInfo := &WorkflowExecution{
		ID:    workflowID,
		RunID: response.GetRunId()}
	return executionInfo, nil
}
func (wc *workflowClient) ExecuteWorkflow(ctx context.Context, options StartWorkflowOptions, workflow interface{}, args ...interface{}) (WorkflowRun, error) {
	var runID string
	var workflowID string
	executionInfo, err := wc.StartWorkflow(ctx, options, workflow, args...)
	if err != nil {
		if alreadyStartedErr, ok := err.(*s.WorkflowExecutionAlreadyStartedError); ok {
			runID = alreadyStartedErr.GetRunId()
			workflowID = options.ID
		} else {
			return nil, err
		}
	} else {
		runID = executionInfo.RunID
		workflowID = executionInfo.ID
	}
	iterFn := func(fnCtx context.Context, fnRunID string) HistoryEventIterator {
		return wc.GetWorkflowHistory(fnCtx, workflowID, fnRunID, true, s.HistoryEventFilterTypeCloseEvent)
	}
	return &workflowRunImpl{
		workflowFn:    workflow,
		workflowID:    workflowID,
		firstRunID:    runID,
		currentRunID:  runID,
		iterFn:        iterFn,
		dataConverter: wc.dataConverter,
		registry:      wc.registry,
	}, nil
}
func (wc *workflowClient) GetWorkflow(ctx context.Context, workflowID string, runID string) WorkflowRun {
	iterFn := func(fnCtx context.Context, fnRunID string) HistoryEventIterator {
		return wc.GetWorkflowHistory(fnCtx, workflowID, fnRunID, true, s.HistoryEventFilterTypeCloseEvent)
	}
	return &workflowRunImpl{
		workflowID:    workflowID,
		firstRunID:    runID,
		currentRunID:  runID,
		iterFn:        iterFn,
		dataConverter: wc.dataConverter,
		registry:      wc.registry,
	}
}
func (wc *workflowClient) SignalWorkflow(ctx context.Context, workflowID string, runID string, signalName string, arg interface{}) error {
	input, err := encodeArg(wc.dataConverter, arg)
	if err != nil {
		return err
	}
	return signalWorkflow(ctx, wc.workflowService, wc.identity, wc.domain, workflowID, runID, signalName, input, wc.featureFlags)
}
func (wc *workflowClient) SignalWithStartWorkflow(ctx context.Context, workflowID string, signalName string, signalArg interface{},
	options StartWorkflowOptions, workflowFunc interface{}, workflowArgs ...interface{}) (*WorkflowExecution, error) {
	signalInput, err := encodeArg(wc.dataConverter, signalArg)
	if err != nil {
		return nil, err
	}
	if workflowID == "" {
		workflowID = uuid.NewRandom().String()
	}
	if options.TaskList == "" {
		return nil, errors.New("missing TaskList")
	}
	executionTimeout := common.Int32Ceil(options.ExecutionStartToCloseTimeout.Seconds())
	if executionTimeout <= 0 {
		return nil, errors.New("missing or invalid ExecutionStartToCloseTimeout")
	}
	decisionTaskTimeout := common.Int32Ceil(options.DecisionTaskStartToCloseTimeout.Seconds())
	if decisionTaskTimeout < 0 {
		return nil, errors.New("negative DecisionTaskStartToCloseTimeout provided")
	}
	if decisionTaskTimeout == 0 {
		decisionTaskTimeout = defaultDecisionTaskTimeoutInSecs
	}
	workflowType, input, err := getValidatedWorkflowFunction(workflowFunc, workflowArgs, wc.dataConverter, wc.registry)
	if err != nil {
		return nil, err
	}
	memo, err := getWorkflowMemo(options.Memo, wc.dataConverter)
	if err != nil {
		return nil, err
	}
	searchAttr, err := serializeSearchAttributes(options.SearchAttributes)
	if err != nil {
		return nil, err
	}
	delayStartSeconds := common.Int32Ceil(options.DelayStart.Seconds())
	if delayStartSeconds < 0 {
		return nil, errors.New("Invalid DelayStart option")
	}
	ctx, span := createOpenTracingWorkflowSpan(ctx, wc.tracer, time.Now(), fmt.Sprintf("SignalWithStartWorkflow-%s", workflowType.Name), workflowID)
	span.Finish()
	header := wc.getWorkflowHeader(ctx)
	signalWithStartRequest := &s.SignalWithStartWorkflowExecutionRequest{
		Domain:                              common.StringPtr(wc.domain),
		RequestId:                           common.StringPtr(uuid.New()),
		WorkflowId:                          common.StringPtr(workflowID),
		WorkflowType:                        workflowTypePtr(*workflowType),
		TaskList:                            common.TaskListPtr(s.TaskList{Name: common.StringPtr(options.TaskList)}),
		Input:                               input,
		ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(executionTimeout),
		TaskStartToCloseTimeoutSeconds:      common.Int32Ptr(decisionTaskTimeout),
		SignalName:                          common.StringPtr(signalName),
		SignalInput:                         signalInput,
		Identity:                            common.StringPtr(wc.identity),
		RetryPolicy:                         convertRetryPolicy(options.RetryPolicy),
		CronSchedule:                        common.StringPtr(options.CronSchedule),
		Memo:                                memo,
		SearchAttributes:                    searchAttr,
		WorkflowIdReusePolicy:               options.WorkflowIDReusePolicy.toThriftPtr(),
		Header:                              header,
		DelayStartSeconds:                   common.Int32Ptr(delayStartSeconds),
	}
	var response *s.StartWorkflowExecutionResponse
	err = backoff.Retry(ctx,
		func() error {
			tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags)
			defer cancel()
			var err1 error
			response, err1 = wc.workflowService.SignalWithStartWorkflowExecution(tchCtx, signalWithStartRequest, opt...)
			return err1
		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
	if err != nil {
		return nil, err
	}
	if wc.metricsScope != nil {
		scope := wc.metricsScope.GetTaggedScope(tagTaskList, options.TaskList, tagWorkflowType, workflowType.Name)
		scope.Counter(metrics.WorkflowSignalWithStartCounter).Inc(1)
	}
	executionInfo := &WorkflowExecution{
		ID:    options.ID,
		RunID: response.GetRunId()}
	return executionInfo, nil
}
func (wc *workflowClient) CancelWorkflow(ctx context.Context, workflowID string, runID string) error {
	request := &s.RequestCancelWorkflowExecutionRequest{
		Domain: common.StringPtr(wc.domain),
		WorkflowExecution: &s.WorkflowExecution{
			WorkflowId: common.StringPtr(workflowID),
			RunId:      getRunID(runID),
		},
		Identity: common.StringPtr(wc.identity),
	}
	return backoff.Retry(ctx,
		func() error {
			tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags)
			defer cancel()
			return wc.workflowService.RequestCancelWorkflowExecution(tchCtx, request, opt...)
		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
}
func (wc *workflowClient) TerminateWorkflow(ctx context.Context, workflowID string, runID string, reason string, details []byte) error {
	request := &s.TerminateWorkflowExecutionRequest{
		Domain: common.StringPtr(wc.domain),
		WorkflowExecution: &s.WorkflowExecution{
			WorkflowId: common.StringPtr(workflowID),
			RunId:      getRunID(runID),
		},
		Reason:   common.StringPtr(reason),
		Details:  details,
		Identity: common.StringPtr(wc.identity),
	}
	err := backoff.Retry(ctx,
		func() error {
			tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags)
			defer cancel()
			return wc.workflowService.TerminateWorkflowExecution(tchCtx, request, opt...)
		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
	return err
}
func (wc *workflowClient) GetWorkflowHistory(
	ctx context.Context,
	workflowID string,
	runID string,
	isLongPoll bool,
	filterType s.HistoryEventFilterType,
) HistoryEventIterator {
	domain := wc.domain
	paginate := func(nextToken []byte) (*s.GetWorkflowExecutionHistoryResponse, error) {
		request := &s.GetWorkflowExecutionHistoryRequest{
			Domain: common.StringPtr(domain),
			Execution: &s.WorkflowExecution{
				WorkflowId: common.StringPtr(workflowID),
				RunId:      getRunID(runID),
			},
			WaitForNewEvent:        common.BoolPtr(isLongPoll),
			HistoryEventFilterType: &filterType,
			NextPageToken:          nextToken,
			SkipArchival:           common.BoolPtr(isLongPoll),
		}
		var response *s.GetWorkflowExecutionHistoryResponse
		var err error
	Loop:
		for {
			err = backoff.Retry(ctx,
				func() error {
					var err1 error
					tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags, func(builder *contextBuilder) {
						if isLongPoll {
							builder.Timeout = defaultGetHistoryTimeoutInSecs * time.Second
						}
					})
					defer cancel()
					response, err1 = wc.workflowService.GetWorkflowExecutionHistory(tchCtx, request, opt...)
					if err1 != nil {
						return err1
					}
					if response.RawHistory != nil {
						history, err := serializer.DeserializeBlobDataToHistoryEvents(response.RawHistory, filterType)
						if err != nil {
							return err
						}
						response.History = history
					}
					return err1
				},
				createDynamicServiceRetryPolicy(ctx),
				func(err error) bool {
					return isServiceTransientError(err) || isEntityNonExistFromPassive(err)
				},
			)
			if err != nil {
				return nil, err
			}
			if isLongPoll && len(response.History.Events) == 0 && len(response.NextPageToken) != 0 {
				request.NextPageToken = response.NextPageToken
				continue Loop
			}
			break Loop
		}
		return response, nil
	}
	return &historyEventIteratorImpl{
		paginate: paginate,
	}
}
func isEntityNonExistFromPassive(err error) bool {
	if nonExistError, ok := err.(*s.EntityNotExistsError); ok {
		return nonExistError.GetActiveCluster() != "" &&
			nonExistError.GetCurrentCluster() != "" &&
			nonExistError.GetActiveCluster() != nonExistError.GetCurrentCluster()
	}
	return false
}