func (wc *workflowClient) ListOpenWorkflow(ctx context.Context, request *s.ListOpenWorkflowExecutionsRequest) (*s.ListOpenWorkflowExecutionsResponse, error) {
	if len(request.GetDomain()) == 0 {
		request.Domain = common.StringPtr(wc.domain)
	}
	var response *s.ListOpenWorkflowExecutionsResponse
	err := backoff.Retry(ctx,
		func() error {
			var err1 error
			tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags)
			defer cancel()
			response, err1 = wc.workflowService.ListOpenWorkflowExecutions(tchCtx, request, opt...)
			return err1
		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
	if err != nil {
		return nil, err
	}
	return response, nil
}
ListOpenWorkflow gets open workflow executions based on request filters
The errors it can throw:
- BadRequestError
- InternalServiceError
- EntityNotExistError
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
}
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)
}
func (wc *workflowClient) CompleteActivityByID(ctx context.Context, domain, workflowID, runID, activityID string,
	result interface{}, err error) error {
	if activityID == "" || workflowID == "" || domain == "" {
		return errors.New("empty activity or workflow id or domainName")
	}
	var data []byte
	if result != nil {
		var err0 error
		data, err0 = encodeArg(wc.dataConverter, result)
		if err0 != nil {
			return err0
		}
	}
	request := convertActivityResultToRespondRequestByID(wc.identity, domain, workflowID, runID, activityID, data, err, wc.dataConverter)
	return reportActivityCompleteByID(ctx, wc.workflowService, request, wc.metricsScope, wc.featureFlags)
}
func (wc *workflowClient) RecordActivityHeartbeat(ctx context.Context, taskToken []byte, details ...interface{}) error {
	data, err := encodeArgs(wc.dataConverter, details)
	if err != nil {
		return err
	}
	return recordActivityHeartbeat(ctx, wc.workflowService, wc.identity, taskToken, data, wc.featureFlags)
}
func (wc *workflowClient) RecordActivityHeartbeatByID(ctx context.Context,
	domain, workflowID, runID, activityID string, details ...interface{}) error {
	data, err := encodeArgs(wc.dataConverter, details)
	if err != nil {
		return err
	}
	return recordActivityHeartbeatByID(ctx, wc.workflowService, wc.identity, domain, workflowID, runID, activityID, data, wc.featureFlags)
}
func (wc *workflowClient) ListClosedWorkflow(ctx context.Context, request *s.ListClosedWorkflowExecutionsRequest) (*s.ListClosedWorkflowExecutionsResponse, error) {
	if len(request.GetDomain()) == 0 {
		request.Domain = common.StringPtr(wc.domain)
	}
	var response *s.ListClosedWorkflowExecutionsResponse
	err := backoff.Retry(ctx,
		func() error {
			var err1 error
			tchCtx, cancel, opt := newChannelContext(ctx, wc.featureFlags)
			defer cancel()
			response, err1 = wc.workflowService.ListClosedWorkflowExecutions(tchCtx, request, opt...)
			return err1
		}, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
	if err != nil {
		return nil, err
	}
	return response, nil
}