func (aw *activityWorker) Stop() {
	select {
	case <-aw.stopC:
	default:
		close(aw.stopC)
	}
	aw.worker.Stop()
}
Shutdown the worker.
package internal
import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"os"
	"reflect"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/opentracing/opentracing-go"
	"github.com/pborman/uuid"
	"github.com/uber-go/tally"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	"go.uber.org/cadence/.gen/go/shared"
	"go.uber.org/cadence/internal/common/auth"
	"go.uber.org/cadence/internal/common/backoff"
	"go.uber.org/cadence/internal/common/metrics"
	"go.uber.org/cadence/internal/common/util"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)
const (
	defaultConcurrentPollRoutineSize = 2
	defaultMaxConcurrentActivityExecutionSize = 1000    
	defaultWorkerActivitiesPerSecond          = 100000  
	defaultMaxConcurrentLocalActivityExecutionSize = 1000    
	defaultWorkerLocalActivitiesPerSecond          = 100000  
	defaultTaskListActivitiesPerSecond = 100000.0  
	defaultMaxConcurrentTaskExecutionSize = 1000    
	defaultWorkerTaskExecutionRate        = 100000  
	defaultPollerRate = 1000
	defaultMaxConcurrentSessionExecutionSize = 1000  
	testTagsContextKey = "cadence-testTags"
)
type (
	workflowWorker struct {
		executionParameters workerExecutionParameters
		workflowService     workflowserviceclient.Interface
		domain              string
		poller              taskPoller  
		worker              *baseWorker
		localActivityWorker *baseWorker
		identity            string
		stopC               chan struct{}
	}
	activityWorker struct {
		executionParameters workerExecutionParameters
		workflowService     workflowserviceclient.Interface
		domain              string
		poller              taskPoller
		worker              *baseWorker
		identity            string
		stopC               chan struct{}
	}
	sessionWorker struct {
		creationWorker *activityWorker
		activityWorker *activityWorker
	}
	workerOverrides struct {
		workflowTaskHandler                WorkflowTaskHandler
		activityTaskHandler                ActivityTaskHandler
		useLocallyDispatchedActivityPoller bool
	}
	workerExecutionParameters struct {
		TaskList string
		ConcurrentActivityExecutionSize int
		WorkerActivitiesPerSecond float64
		MaxConcurrentActivityPollers int
		ConcurrentDecisionTaskExecutionSize int
		WorkerDecisionTasksPerSecond float64
		MaxConcurrentDecisionPollers int
		ConcurrentLocalActivityExecutionSize int
		WorkerLocalActivitiesPerSecond float64
		TaskListActivitiesPerSecond float64
		Identity string
		MetricsScope tally.Scope
		Logger *zap.Logger
		EnableLoggingInReplay bool
		UserContext context.Context
		UserContextCancel context.CancelFunc
		DisableStickyExecution bool
		StickyScheduleToStartTimeout time.Duration
		NonDeterministicWorkflowPolicy NonDeterministicWorkflowPolicy
		DataConverter DataConverter
		WorkerStopTimeout time.Duration
		WorkerStopChannel <-chan struct{}
		SessionResourceID string
		ContextPropagators []ContextPropagator
		Tracer opentracing.Tracer
		WorkflowInterceptors []WorkflowInterceptorFactory
		FeatureFlags FeatureFlags
	}
)
func newWorkflowWorker(
	service workflowserviceclient.Interface,
	domain string,
	params workerExecutionParameters,
	ppMgr pressurePointMgr,
	registry *registry,
	ldaTunnel *locallyDispatchedActivityTunnel,
) *workflowWorker {
	return newWorkflowWorkerInternal(service, domain, params, ppMgr, nil, registry, ldaTunnel)
}
func ensureRequiredParams(params *workerExecutionParameters) {
	if params.Identity == "" {
		params.Identity = getWorkerIdentity(params.TaskList)
	}
	if params.Logger == nil {
		config := zap.NewProductionConfig()
		config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
		logger, _ := config.Build()
		params.Logger = logger
		params.Logger.Info("No logger configured for cadence worker. Created default one.")
	}
	if params.MetricsScope == nil {
		params.MetricsScope = tally.NoopScope
		params.Logger.Info("No metrics scope configured for cadence worker. Use NoopScope as default.")
	}
	if params.DataConverter == nil {
		params.DataConverter = getDefaultDataConverter()
		params.Logger.Info("No DataConverter configured for cadence worker. Use default one.")
	}
	if params.UserContext == nil {
		params.UserContext = context.Background()
	}
}
func verifyDomainExist(
	client workflowserviceclient.Interface,
	domain string,
	logger *zap.Logger,
	featureFlags FeatureFlags,
) error {
	ctx := context.Background()
	descDomainOp := func() error {
		tchCtx, cancel, opt := newChannelContext(ctx, featureFlags)
		defer cancel()
		_, err := client.DescribeDomain(tchCtx, &shared.DescribeDomainRequest{Name: &domain}, opt...)
		if err != nil {
			if _, ok := err.(*shared.EntityNotExistsError); ok {
				logger.Error("domain does not exist", zap.String("domain", domain), zap.Error(err))
				return err
			}
			if _, ok := err.(*shared.BadRequestError); ok {
				logger.Error("domain does not exist", zap.String("domain", domain), zap.Error(err))
				return err
			}
			logger.Warn("unable to verify if domain exist", zap.String("domain", domain), zap.Error(err))
		}
		return nil
	}
	if len(domain) == 0 {
		return errors.New("domain cannot be empty")
	}
	return backoff.Retry(ctx, descDomainOp, createDynamicServiceRetryPolicy(ctx), isServiceTransientError)
}
func newWorkflowWorkerInternal(
	service workflowserviceclient.Interface,
	domain string,
	params workerExecutionParameters,
	ppMgr pressurePointMgr,
	overrides *workerOverrides,
	registry *registry,
	ldaTunnel *locallyDispatchedActivityTunnel,
) *workflowWorker {
	workerStopChannel := make(chan struct{})
	params.WorkerStopChannel = getReadOnlyChannel(workerStopChannel)
	ensureRequiredParams(&params)
	var taskHandler WorkflowTaskHandler
	if overrides != nil && overrides.workflowTaskHandler != nil {
		taskHandler = overrides.workflowTaskHandler
	} else {
		taskHandler = newWorkflowTaskHandler(domain, params, ppMgr, registry)
	}
	return newWorkflowTaskWorkerInternal(taskHandler, service, domain, params, workerStopChannel, ldaTunnel)
}
func newWorkflowTaskWorkerInternal(
	taskHandler WorkflowTaskHandler,
	service workflowserviceclient.Interface,
	domain string,
	params workerExecutionParameters,
	stopC chan struct{},
	ldaTunnel *locallyDispatchedActivityTunnel,
) *workflowWorker {
	ensureRequiredParams(&params)
	poller := newWorkflowTaskPoller(
		taskHandler,
		ldaTunnel,
		service,
		domain,
		params,
	)
	worker := newBaseWorker(baseWorkerOptions{
		pollerCount:       params.MaxConcurrentDecisionPollers,
		pollerRate:        defaultPollerRate,
		maxConcurrentTask: params.ConcurrentDecisionTaskExecutionSize,
		maxTaskPerSecond:  params.WorkerDecisionTasksPerSecond,
		taskWorker:        poller,
		identity:          params.Identity,
		workerType:        "DecisionWorker",
		shutdownTimeout:   params.WorkerStopTimeout},
		params.Logger,
		params.MetricsScope,
		nil,
	)
	laTunnel := newLocalActivityTunnel(params.WorkerStopChannel)
	if handlerImpl, ok := taskHandler.(*workflowTaskHandlerImpl); ok {
		handlerImpl.laTunnel = laTunnel
	}
	localActivityTaskPoller := newLocalActivityPoller(params, laTunnel)
	localActivityWorker := newBaseWorker(baseWorkerOptions{
		pollerCount:       1,  
		maxConcurrentTask: params.ConcurrentLocalActivityExecutionSize,
		maxTaskPerSecond:  params.WorkerLocalActivitiesPerSecond,
		taskWorker:        localActivityTaskPoller,
		identity:          params.Identity,
		workerType:        "LocalActivityWorker",
		shutdownTimeout:   params.WorkerStopTimeout},
		params.Logger,
		params.MetricsScope,
		nil,
	)
	worker.taskQueueCh = laTunnel.resultCh
	return &workflowWorker{
		executionParameters: params,
		workflowService:     service,
		poller:              poller,
		worker:              worker,
		localActivityWorker: localActivityWorker,
		identity:            params.Identity,
		domain:              domain,
		stopC:               stopC,
	}
}
func (ww *workflowWorker) Start() error {
	err := verifyDomainExist(ww.workflowService, ww.domain, ww.worker.logger, ww.executionParameters.FeatureFlags)
	if err != nil {
		return err
	}
	ww.localActivityWorker.Start()
	ww.worker.Start()
	return nil  
}
func (ww *workflowWorker) Run() error {
	err := verifyDomainExist(ww.workflowService, ww.domain, ww.worker.logger, ww.executionParameters.FeatureFlags)
	if err != nil {
		return err
	}
	ww.localActivityWorker.Start()
	ww.worker.Run()
	return nil
}
func (ww *workflowWorker) Stop() {
	select {
	case <-ww.stopC:
	default:
		close(ww.stopC)
	}
	ww.localActivityWorker.Stop()
	ww.worker.Stop()
}
func newSessionWorker(service workflowserviceclient.Interface,
	domain string,
	params workerExecutionParameters,
	overrides *workerOverrides,
	env *registry,
	maxConcurrentSessionExecutionSize int,
) *sessionWorker {
	ensureRequiredParams(&params)
	if params.SessionResourceID == "" {
		params.SessionResourceID = uuid.New()
	}
	sessionEnvironment := newSessionEnvironment(params.SessionResourceID, maxConcurrentSessionExecutionSize)
	creationTasklist := getCreationTasklist(params.TaskList)
	params.UserContext = context.WithValue(params.UserContext, sessionEnvironmentContextKey, sessionEnvironment)
	params.TaskList = sessionEnvironment.GetResourceSpecificTasklist()
	activityWorker := newActivityWorker(service, domain, params, overrides, env, nil)
	params.MaxConcurrentActivityPollers = 1
	params.TaskList = creationTasklist
	creationWorker := newActivityWorker(service, domain, params, overrides, env, sessionEnvironment.GetTokenBucket())
	return &sessionWorker{
		creationWorker: creationWorker,
		activityWorker: activityWorker,
	}
}
func (sw *sessionWorker) Start() error {
	err := sw.creationWorker.Start()
	if err != nil {
		return err
	}
	err = sw.activityWorker.Start()
	if err != nil {
		sw.creationWorker.Stop()
		return err
	}
	return nil
}
func (sw *sessionWorker) Run() error {
	err := sw.creationWorker.Start()
	if err != nil {
		return err
	}
	return sw.activityWorker.Run()
}
func (sw *sessionWorker) Stop() {
	sw.creationWorker.Stop()
	sw.activityWorker.Stop()
}
func newActivityWorker(
	service workflowserviceclient.Interface,
	domain string,
	params workerExecutionParameters,
	overrides *workerOverrides,
	env *registry,
	sessionTokenBucket *sessionTokenBucket,
) *activityWorker {
	workerStopChannel := make(chan struct{}, 1)
	params.WorkerStopChannel = getReadOnlyChannel(workerStopChannel)
	ensureRequiredParams(&params)
	workerType := "ActivityWorker"
	var taskHandler ActivityTaskHandler
	if overrides != nil && overrides.activityTaskHandler != nil {
		taskHandler = overrides.activityTaskHandler
	} else {
		taskHandler = newActivityTaskHandler(service, params, env)
	}
	var taskPoller taskPoller
	if overrides != nil && overrides.useLocallyDispatchedActivityPoller {
		taskPoller = newLocallyDispatchedActivityTaskPoller(taskHandler, service, domain, params)
		workerType = "LocallyDispatchedActivityWorker"
	} else {
		taskPoller = newActivityTaskPoller(
			taskHandler,
			service,
			domain,
			params,
		)
	}
	return newActivityTaskWorker(service, domain, params, sessionTokenBucket, workerStopChannel, taskPoller, workerType)
}
func newActivityTaskWorker(
	service workflowserviceclient.Interface,
	domain string,
	workerParams workerExecutionParameters,
	sessionTokenBucket *sessionTokenBucket,
	stopC chan struct{},
	poller taskPoller,
	workerType string,
) (worker *activityWorker) {
	ensureRequiredParams(&workerParams)
	base := newBaseWorker(
		baseWorkerOptions{
			pollerCount:       workerParams.MaxConcurrentActivityPollers,
			pollerRate:        defaultPollerRate,
			maxConcurrentTask: workerParams.ConcurrentActivityExecutionSize,
			maxTaskPerSecond:  workerParams.WorkerActivitiesPerSecond,
			taskWorker:        poller,
			identity:          workerParams.Identity,
			workerType:        workerType,
			shutdownTimeout:   workerParams.WorkerStopTimeout,
			userContextCancel: workerParams.UserContextCancel},
		workerParams.Logger,
		workerParams.MetricsScope,
		sessionTokenBucket,
	)
	return &activityWorker{
		executionParameters: workerParams,
		workflowService:     service,
		worker:              base,
		poller:              poller,
		identity:            workerParams.Identity,
		domain:              domain,
		stopC:               stopC,
	}
}
func (aw *activityWorker) Start() error {
	err := verifyDomainExist(aw.workflowService, aw.domain, aw.worker.logger, aw.executionParameters.FeatureFlags)
	if err != nil {
		return err
	}
	aw.worker.Start()
	return nil  
}
func (aw *activityWorker) Run() error {
	err := verifyDomainExist(aw.workflowService, aw.domain, aw.worker.logger, aw.executionParameters.FeatureFlags)
	if err != nil {
		return err
	}
	aw.worker.Run()
	return nil
}