queues

package
v1.17.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 17, 2022 License: MIT Imports: 36 Imported by: 0

Documentation

Overview

Package queues is a generated GoMock package.

Package queues is a generated GoMock package.

Package queues is a generated GoMock package.

Package queues is a generated GoMock package.

Index

Constants

View Source 
const (
	TaskRequests                = "task_requests"
	TaskLatency                 = "task_latency"
	TaskFailures                = "task_errors"
	TaskDiscarded               = "task_errors_discarded"
	TaskSkipped                 = "task_skipped"
	TaskAttempt                 = "task_attempt"
	TaskStandbyRetryCounter     = "task_errors_standby_retry_counter"
	TaskWorkflowBusyCounter     = "task_errors_workflow_busy"
	TaskNotActiveCounter        = "task_errors_not_active_counter"
	TaskProcessingLatency       = "task_latency_processing"
	TaskNoUserProcessingLatency = "task_latency_processing_nouserlatency"
	TaskQueueLatency            = "task_latency_queue"
	TaskNoUserLatency           = "task_latency_nouserlatency"
	TaskUserLatency             = "task_latency_userlatency"
	TaskNoUserQueueLatency      = "task_latency_queue_nouserlatency"
	TaskReschedulerPendingTasks = "task_rescheduler_pending_tasks"
	TaskThrottledCounter        = "task_throttled_counter"

	TaskBatchCompleteCounter    = "task_batch_complete_counter"
	NewTimerNotifyCounter       = "new_timer_notifications"
	AckLevelUpdateCounter       = "ack_level_update"
	AckLevelUpdateFailedCounter = "ack_level_update_failed"
	PendingTasksCounter         = "pending_tasks"
)

Metric names

View Source 
const (
	OperationTimerQueueProcessor           = "TimerQueueProcessor"
	OperationTimerActiveQueueProcessor     = "TimerActiveQueueProcessor"
	OperationTimerStandbyQueueProcessor    = "TimerStandbyQueueProcessor"
	OperationTransferQueueProcessor        = "TransferQueueProcessor"
	OperationTransferActiveQueueProcessor  = "TransferActiveQueueProcessor"
	OperationTransferStandbyQueueProcessor = "TransferStandbyQueueProcessor"
	OperationVisibilityQueueProcessor      = "VisibilityQueueProcessor"
	OperationTaskPriorityAssigner          = "TaskPriorityAssigner"
)

Operation tag value for queue processors

View Source 
const (
	TaskTypeTransferActiveTaskActivity             = "TransferActiveTaskActivity"
	TaskTypeTransferActiveTaskWorkflowTask         = "TransferActiveTaskWorkflowTask"
	TaskTypeTransferActiveTaskCloseExecution       = "TransferActiveTaskCloseExecution"
	TaskTypeTransferActiveTaskCancelExecution      = "TransferActiveTaskCancelExecution"
	TaskTypeTransferActiveTaskSignalExecution      = "TransferActiveTaskSignalExecution"
	TaskTypeTransferActiveTaskStartChildExecution  = "TransferActiveTaskStartChildExecution"
	TaskTypeTransferActiveTaskResetWorkflow        = "TransferActiveTaskResetWorkflow"
	TaskTypeTransferStandbyTaskActivity            = "TransferStandbyTaskActivity"
	TaskTypeTransferStandbyTaskWorkflowTask        = "TransferStandbyTaskWorkflowTask"
	TaskTypeTransferStandbyTaskCloseExecution      = "TransferStandbyTaskCloseExecution"
	TaskTypeTransferStandbyTaskCancelExecution     = "TransferStandbyTaskCancelExecution"
	TaskTypeTransferStandbyTaskSignalExecution     = "TransferStandbyTaskSignalExecution"
	TaskTypeTransferStandbyTaskStartChildExecution = "TransferStandbyTaskStartChildExecution"
	TaskTypeTransferStandbyTaskResetWorkflow       = "TransferStandbyTaskResetWorkflow"
	TaskTypeVisibilityTaskStartExecution           = "VisibilityTaskStartExecution"
	TaskTypeVisibilityTaskUpsertExecution          = "VisibilityTaskUpsertExecution"
	TaskTypeVisibilityTaskCloseExecution           = "VisibilityTaskCloseExecution"
	TaskTypeVisibilityTaskDeleteExecution          = "VisibilityTaskDeleteExecution"
	TaskTypeTimerActiveTaskActivityTimeout         = "TimerActiveTaskActivityTimeout"
	TaskTypeTimerActiveTaskWorkflowTaskTimeout     = "TimerActiveTaskWorkflowTaskTimeout"
	TaskTypeTimerActiveTaskUserTimer               = "TimerActiveTaskUserTimer"
	TaskTypeTimerActiveTaskWorkflowTimeout         = "TimerActiveTaskWorkflowTimeout"
	TaskTypeTimerActiveTaskActivityRetryTimer      = "TimerActiveTaskActivityRetryTimer"
	TaskTypeTimerActiveTaskWorkflowBackoffTimer    = "TimerActiveTaskWorkflowBackoffTimer"
	TaskTypeTimerActiveTaskDeleteHistoryEvent      = "TimerActiveTaskDeleteHistoryEvent"
	TaskTypeTimerStandbyTaskActivityTimeout        = "TimerStandbyTaskActivityTimeout"
	TaskTypeTimerStandbyTaskWorkflowTaskTimeout    = "TimerStandbyTaskWorkflowTaskTimeout"
	TaskTypeTimerStandbyTaskUserTimer              = "TimerStandbyTaskUserTimer"
	TaskTypeTimerStandbyTaskWorkflowTimeout        = "TimerStandbyTaskWorkflowTimeout"
	TaskTypeTimerStandbyTaskActivityRetryTimer     = "TimerStandbyTaskActivityRetryTimer"
	TaskTypeTimerStandbyTaskWorkflowBackoffTimer   = "TimerStandbyTaskWorkflowBackoffTimer"
	TaskTypeTimerStandbyTaskDeleteHistoryEvent     = "TimerStandbyTaskDeleteHistoryEvent"
)

Task type tag value for active and standby tasks

View Source 
const (
	FactoryFxGroup = "queueFactory"
)

Variables

This section is empty.

Functions

func FromPersistenceAndPredicate added in v1.17.3 

func FromPersistenceAndPredicate(
	attributes *persistencespb.AndPredicateAttributes,
) tasks.Predicate

func FromPersistenceEmptyPredicate added in v1.17.3 

func FromPersistenceEmptyPredicate(
	_ *persistencespb.EmptyPredicateAttributes,
) tasks.Predicate

func FromPersistenceNamespaceIDPredicate added in v1.17.3 

func FromPersistenceNamespaceIDPredicate(
	attributes *persistencespb.NamespaceIdPredicateAttributes,
) tasks.Predicate

func FromPersistenceNotPredicate added in v1.17.3 

func FromPersistenceNotPredicate(
	attributes *persistencespb.NotPredicateAttributes,
) tasks.Predicate

func FromPersistenceOrPredicate added in v1.17.3 

func FromPersistenceOrPredicate(
	attributes *persistencespb.OrPredicateAttributes,
) tasks.Predicate

func FromPersistencePredicate added in v1.17.3 

func FromPersistencePredicate(
	predicate *persistencespb.Predicate,
) tasks.Predicate

func FromPersistenceQueueState added in v1.17.3 

func FromPersistenceQueueState(
	state *persistencespb.QueueState,
) *queueState

func FromPersistenceTaskKey added in v1.17.3 

func FromPersistenceTaskKey(
	key *persistencespb.TaskKey,
) tasks.Key

func FromPersistenceTaskTypePredicate added in v1.17.3 

func FromPersistenceTaskTypePredicate(
	attributes *persistencespb.TaskTypePredicateAttributes,
) tasks.Predicate

func FromPersistenceUniversalPredicate added in v1.17.3 

func FromPersistenceUniversalPredicate(
	_ *persistencespb.UniversalPredicateAttributes,
) tasks.Predicate

func GetActiveTimerTaskTypeTagValue added in v1.17.0 

func GetActiveTimerTaskTypeTagValue(
	task tasks.Task,
) string

func GetActiveTransferTaskTypeTagValue added in v1.17.0 

func GetActiveTransferTaskTypeTagValue(
	task tasks.Task,
) string

func GetStandbyTimerTaskTypeTagValue added in v1.17.0 

func GetStandbyTimerTaskTypeTagValue(
	task tasks.Task,
) string

func GetStandbyTransferTaskTypeTagValue added in v1.17.0 

func GetStandbyTransferTaskTypeTagValue(
	task tasks.Task,
) string

func GetVisibilityTaskTypeTagValue added in v1.17.0 

func GetVisibilityTaskTypeTagValue(
	task tasks.Task,
) string

func NewImmediateQueue added in v1.17.3 

func NewImmediateQueue(
	shard shard.Context,
	category tasks.Category,
	scheduler Scheduler,
	executor Executor,
	options *Options,
	rateLimiter quotas.RateLimiter,
	logger log.Logger,
	metricsHandler metrics.MetricsHandler,
) *immediateQueue

func NewRandomKey added in v1.17.0 

func NewRandomKey() tasks.Key

func NewRandomKeyInRange added in v1.17.0 

func NewRandomKeyInRange(
	r Range,
) tasks.Key

func NewRescheduler added in v1.17.0 

func NewRescheduler(
	scheduler Scheduler,
	timeSource clock.TimeSource,
	logger log.Logger,
	metricsHandler metrics.MetricsHandler,
) *reschedulerImpl

func NewScheduledQueue added in v1.17.3 

func NewScheduledQueue(
	shard shard.Context,
	category tasks.Category,
	scheduler Scheduler,
	executor Executor,
	options *Options,
	rateLimiter quotas.RateLimiter,
	logger log.Logger,
	metricsHandler metrics.MetricsHandler,
) *scheduledQueue

func NewScheduler added in v1.17.0 

func NewScheduler(
	priorityAssigner PriorityAssigner,
	options SchedulerOptions,
	metricsProvider metrics.MetricsHandler,
	logger log.Logger,
) *schedulerImpl

func ToPersistenceAndPredicate added in v1.17.3 

func ToPersistenceAndPredicate(
	andPredicate *predicates.AndImpl[tasks.Task],
) *persistencespb.Predicate

func ToPersistenceEmptyPredicate added in v1.17.3 

func ToPersistenceEmptyPredicate(
	_ *predicates.EmptyImpl[tasks.Task],
) *persistencespb.Predicate

func ToPersistenceNamespaceIDPredicate added in v1.17.3 

func ToPersistenceNamespaceIDPredicate(
	namespaceIDPredicate *tasks.NamespacePredicate,
) *persistencespb.Predicate

func ToPersistenceNotPredicate added in v1.17.3 

func ToPersistenceNotPredicate(
	notPredicate *predicates.NotImpl[tasks.Task],
) *persistencespb.Predicate

func ToPersistenceOrPredicate added in v1.17.3 

func ToPersistenceOrPredicate(
	orPredicate *predicates.OrImpl[tasks.Task],
) *persistencespb.Predicate

func ToPersistencePredicate added in v1.17.3 

func ToPersistencePredicate(
	predicate tasks.Predicate,
) *persistencespb.Predicate

func ToPersistenceQueueState added in v1.17.3 

func ToPersistenceQueueState(
	queueState *queueState,
) *persistencespb.QueueState

func ToPersistenceRange added in v1.17.3 

func ToPersistenceRange(
	r Range,
) *persistencespb.QueueSliceRange

func ToPersistenceScope added in v1.17.3 

func ToPersistenceScope(
	scope Scope,
) *persistencespb.QueueSliceScope

func ToPersistenceTaskKey added in v1.17.3 

func ToPersistenceTaskKey(
	key tasks.Key,
) *persistencespb.TaskKey

func ToPersistenceTaskTypePredicate added in v1.17.3 

func ToPersistenceTaskTypePredicate(
	taskTypePredicate *tasks.TypePredicate,
) *persistencespb.Predicate

func ToPersistenceUniversalPredicate added in v1.17.3 

func ToPersistenceUniversalPredicate(
	_ *predicates.UniversalImpl[tasks.Task],
) *persistencespb.Predicate

Types

type Executable added in v1.17.0 

type Executable interface {
	ctasks.PriorityTask
	tasks.Task

	Attempt() int
	Logger() log.Logger
	GetTask() tasks.Task

	QueueType() QueueType
}

func NewExecutable added in v1.17.0 

func NewExecutable(
	task tasks.Task,
	filter TaskFilter,
	executor Executor,
	scheduler Scheduler,
	rescheduler Rescheduler,
	timeSource clock.TimeSource,
	namespaceRegistry namespace.Registry,
	logger log.Logger,
	criticalRetryAttempt dynamicconfig.IntPropertyFn,
	queueType QueueType,
	namespaceCacheRefreshInterval dynamicconfig.DurationPropertyFn,
) Executable

type ExecutableInitializer added in v1.17.3 

type ExecutableInitializer func(tasks.Task) Executable

type Executor added in v1.17.0 

type Executor interface {
	Execute(context.Context, Executable) (metrics.MetricsHandler, error)
}

func NewExecutorWrapper added in v1.17.0 

func NewExecutorWrapper(
	currentClusterName string,
	registry namespace.Registry,
	activeExecutor Executor,
	standbyExecutor Executor,
	logger log.Logger,
) Executor

type Factory added in v1.17.3 

type Factory interface {
	common.Daemon

	// TODO: remove the cache parameter after workflow cache become a host level component
	// and it can be provided as a parameter when creating a QueueFactory instance.
	// Currently, workflow cache is shard level, but we can't get it from shard or engine interface,
	// as that will lead to a cycle dependency issue between shard and workflow package.
	CreateQueue(shard shard.Context, engine shard.Engine, cache workflow.Cache) Queue
}

type Iterator added in v1.17.0 

type Iterator interface {
	collection.Iterator[tasks.Task]

	Range() Range
	CanSplit(tasks.Key) bool
	Split(key tasks.Key) (left Iterator, right Iterator)
	CanMerge(Iterator) bool
	Merge(Iterator) Iterator
	Remaining() Iterator
}

type IteratorImpl added in v1.17.0 

type IteratorImpl struct {
	// contains filtered or unexported fields
}

func NewIterator added in v1.17.0 

func NewIterator(
	paginationFnProvider PaginationFnProvider,
	r Range,
) *IteratorImpl

func (*IteratorImpl) CanMerge added in v1.17.0 

func (i *IteratorImpl) CanMerge(iter Iterator) bool

func (*IteratorImpl) CanSplit added in v1.17.0 

func (i *IteratorImpl) CanSplit(key tasks.Key) bool

func (*IteratorImpl) HasNext added in v1.17.0 

func (i *IteratorImpl) HasNext() bool

func (*IteratorImpl) Merge added in v1.17.0 

func (i *IteratorImpl) Merge(iter Iterator) Iterator

func (*IteratorImpl) Next added in v1.17.0 

func (i *IteratorImpl) Next() (tasks.Task, error)

func (*IteratorImpl) Range added in v1.17.0 

func (i *IteratorImpl) Range() Range

func (*IteratorImpl) Remaining added in v1.17.0 

func (i *IteratorImpl) Remaining() Iterator

func (*IteratorImpl) Split added in v1.17.0 

func (i *IteratorImpl) Split(key tasks.Key) (left Iterator, right Iterator)

type MockExecutable added in v1.17.0 

type MockExecutable struct {
	// contains filtered or unexported fields
}

MockExecutable is a mock of Executable interface.

func NewMockExecutable added in v1.17.0 

func NewMockExecutable(ctrl *gomock.Controller) *MockExecutable

NewMockExecutable creates a new mock instance.

func (*MockExecutable) Ack added in v1.17.0 

func (m *MockExecutable) Ack()

Ack mocks base method.

func (*MockExecutable) Attempt added in v1.17.0 

func (m *MockExecutable) Attempt() int

Attempt mocks base method.

func (*MockExecutable) Cancel added in v1.17.3 

func (m *MockExecutable) Cancel()

Cancel mocks base method.

func (*MockExecutable) EXPECT added in v1.17.0 

func (m *MockExecutable) EXPECT() *MockExecutableMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockExecutable) Execute added in v1.17.0 

func (m *MockExecutable) Execute() error

Execute mocks base method.

func (*MockExecutable) GetCategory added in v1.17.0 

func (m *MockExecutable) GetCategory() tasks0.Category

GetCategory mocks base method.

func (*MockExecutable) GetKey added in v1.17.0 

func (m *MockExecutable) GetKey() tasks0.Key

GetKey mocks base method.

func (*MockExecutable) GetNamespaceID added in v1.17.0 

func (m *MockExecutable) GetNamespaceID() string

GetNamespaceID mocks base method.

func (*MockExecutable) GetPriority added in v1.17.0 

func (m *MockExecutable) GetPriority() tasks.Priority

GetPriority mocks base method.

func (*MockExecutable) GetRunID added in v1.17.0 

func (m *MockExecutable) GetRunID() string

GetRunID mocks base method.

func (*MockExecutable) GetTask added in v1.17.0 

func (m *MockExecutable) GetTask() tasks0.Task

GetTask mocks base method.

func (*MockExecutable) GetTaskID added in v1.17.0 

func (m *MockExecutable) GetTaskID() int64

GetTaskID mocks base method.

func (*MockExecutable) GetType added in v1.17.0 

func (m *MockExecutable) GetType() v1.TaskType

GetType mocks base method.

func (*MockExecutable) GetVersion added in v1.17.0 

func (m *MockExecutable) GetVersion() int64

GetVersion mocks base method.

func (*MockExecutable) GetVisibilityTime added in v1.17.0 

func (m *MockExecutable) GetVisibilityTime() time.Time

GetVisibilityTime mocks base method.

func (*MockExecutable) GetWorkflowID added in v1.17.0 

func (m *MockExecutable) GetWorkflowID() string

GetWorkflowID mocks base method.

func (*MockExecutable) HandleErr added in v1.17.0 

func (m *MockExecutable) HandleErr(err error) error

HandleErr mocks base method.

func (*MockExecutable) IsRetryableError added in v1.17.0 

func (m *MockExecutable) IsRetryableError(err error) bool

IsRetryableError mocks base method.

func (*MockExecutable) Logger added in v1.17.0 

func (m *MockExecutable) Logger() log.Logger

Logger mocks base method.

func (*MockExecutable) Nack added in v1.17.0 

func (m *MockExecutable) Nack(err error)

Nack mocks base method.

func (*MockExecutable) QueueType added in v1.17.0 

func (m *MockExecutable) QueueType() QueueType

QueueType mocks base method.

func (*MockExecutable) Reschedule added in v1.17.0 

func (m *MockExecutable) Reschedule()

Reschedule mocks base method.

func (*MockExecutable) RetryPolicy added in v1.17.0 

func (m *MockExecutable) RetryPolicy() backoff.RetryPolicy

RetryPolicy mocks base method.

func (*MockExecutable) SetPriority added in v1.17.0 

func (m *MockExecutable) SetPriority(arg0 tasks.Priority)

SetPriority mocks base method.

func (*MockExecutable) SetTaskID added in v1.17.0 

func (m *MockExecutable) SetTaskID(id int64)

SetTaskID mocks base method.

func (*MockExecutable) SetVersion added in v1.17.0 

func (m *MockExecutable) SetVersion(version int64)

SetVersion mocks base method.

func (*MockExecutable) SetVisibilityTime added in v1.17.0 

func (m *MockExecutable) SetVisibilityTime(timestamp time.Time)

SetVisibilityTime mocks base method.

func (*MockExecutable) State added in v1.17.0 

func (m *MockExecutable) State() tasks.State

State mocks base method.

type MockExecutableMockRecorder added in v1.17.0 

type MockExecutableMockRecorder struct {
	// contains filtered or unexported fields
}

MockExecutableMockRecorder is the mock recorder for MockExecutable.

func (*MockExecutableMockRecorder) Ack added in v1.17.0 

func (mr *MockExecutableMockRecorder) Ack() *gomock.Call

Ack indicates an expected call of Ack.

func (*MockExecutableMockRecorder) Attempt added in v1.17.0 

func (mr *MockExecutableMockRecorder) Attempt() *gomock.Call

Attempt indicates an expected call of Attempt.

func (*MockExecutableMockRecorder) Cancel added in v1.17.3 

func (mr *MockExecutableMockRecorder) Cancel() *gomock.Call

Cancel indicates an expected call of Cancel.

func (*MockExecutableMockRecorder) Execute added in v1.17.0 

func (mr *MockExecutableMockRecorder) Execute() *gomock.Call

Execute indicates an expected call of Execute.

func (*MockExecutableMockRecorder) GetCategory added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetCategory() *gomock.Call

GetCategory indicates an expected call of GetCategory.

func (*MockExecutableMockRecorder) GetKey added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetKey() *gomock.Call

GetKey indicates an expected call of GetKey.

func (*MockExecutableMockRecorder) GetNamespaceID added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetNamespaceID() *gomock.Call

GetNamespaceID indicates an expected call of GetNamespaceID.

func (*MockExecutableMockRecorder) GetPriority added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetPriority() *gomock.Call

GetPriority indicates an expected call of GetPriority.

func (*MockExecutableMockRecorder) GetRunID added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetRunID() *gomock.Call

GetRunID indicates an expected call of GetRunID.

func (*MockExecutableMockRecorder) GetTask added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetTask() *gomock.Call

GetTask indicates an expected call of GetTask.

func (*MockExecutableMockRecorder) GetTaskID added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetTaskID() *gomock.Call

GetTaskID indicates an expected call of GetTaskID.

func (*MockExecutableMockRecorder) GetType added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetType() *gomock.Call

GetType indicates an expected call of GetType.

func (*MockExecutableMockRecorder) GetVersion added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetVersion() *gomock.Call

GetVersion indicates an expected call of GetVersion.

func (*MockExecutableMockRecorder) GetVisibilityTime added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetVisibilityTime() *gomock.Call

GetVisibilityTime indicates an expected call of GetVisibilityTime.

func (*MockExecutableMockRecorder) GetWorkflowID added in v1.17.0 

func (mr *MockExecutableMockRecorder) GetWorkflowID() *gomock.Call

GetWorkflowID indicates an expected call of GetWorkflowID.

func (*MockExecutableMockRecorder) HandleErr added in v1.17.0 

func (mr *MockExecutableMockRecorder) HandleErr(err interface{}) *gomock.Call

HandleErr indicates an expected call of HandleErr.

func (*MockExecutableMockRecorder) IsRetryableError added in v1.17.0 

func (mr *MockExecutableMockRecorder) IsRetryableError(err interface{}) *gomock.Call

IsRetryableError indicates an expected call of IsRetryableError.

func (*MockExecutableMockRecorder) Logger added in v1.17.0 

func (mr *MockExecutableMockRecorder) Logger() *gomock.Call

Logger indicates an expected call of Logger.

func (*MockExecutableMockRecorder) Nack added in v1.17.0 

func (mr *MockExecutableMockRecorder) Nack(err interface{}) *gomock.Call

Nack indicates an expected call of Nack.

func (*MockExecutableMockRecorder) QueueType added in v1.17.0 

func (mr *MockExecutableMockRecorder) QueueType() *gomock.Call

QueueType indicates an expected call of QueueType.

func (*MockExecutableMockRecorder) Reschedule added in v1.17.0 

func (mr *MockExecutableMockRecorder) Reschedule() *gomock.Call

Reschedule indicates an expected call of Reschedule.

func (*MockExecutableMockRecorder) RetryPolicy added in v1.17.0 

func (mr *MockExecutableMockRecorder) RetryPolicy() *gomock.Call

RetryPolicy indicates an expected call of RetryPolicy.

func (*MockExecutableMockRecorder) SetPriority added in v1.17.0 

func (mr *MockExecutableMockRecorder) SetPriority(arg0 interface{}) *gomock.Call

SetPriority indicates an expected call of SetPriority.

func (*MockExecutableMockRecorder) SetTaskID added in v1.17.0 

func (mr *MockExecutableMockRecorder) SetTaskID(id interface{}) *gomock.Call

SetTaskID indicates an expected call of SetTaskID.

func (*MockExecutableMockRecorder) SetVersion added in v1.17.0 

func (mr *MockExecutableMockRecorder) SetVersion(version interface{}) *gomock.Call

SetVersion indicates an expected call of SetVersion.

func (*MockExecutableMockRecorder) SetVisibilityTime added in v1.17.0 

func (mr *MockExecutableMockRecorder) SetVisibilityTime(timestamp interface{}) *gomock.Call

SetVisibilityTime indicates an expected call of SetVisibilityTime.

func (*MockExecutableMockRecorder) State added in v1.17.0 

func (mr *MockExecutableMockRecorder) State() *gomock.Call

State indicates an expected call of State.

type MockExecutor added in v1.17.0 

type MockExecutor struct {
	// contains filtered or unexported fields
}

MockExecutor is a mock of Executor interface.

func NewMockExecutor added in v1.17.0 

func NewMockExecutor(ctrl *gomock.Controller) *MockExecutor

NewMockExecutor creates a new mock instance.

func (*MockExecutor) EXPECT added in v1.17.0 

func (m *MockExecutor) EXPECT() *MockExecutorMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockExecutor) Execute added in v1.17.0 

func (m *MockExecutor) Execute(arg0 context.Context, arg1 Executable) (metrics.MetricsHandler, error)

Execute mocks base method.

type MockExecutorMockRecorder added in v1.17.0 

type MockExecutorMockRecorder struct {
	// contains filtered or unexported fields
}

MockExecutorMockRecorder is the mock recorder for MockExecutor.

func (*MockExecutorMockRecorder) Execute added in v1.17.0 

func (mr *MockExecutorMockRecorder) Execute(arg0, arg1 interface{}) *gomock.Call

Execute indicates an expected call of Execute.

type MockFactory added in v1.17.3 

type MockFactory struct {
	// contains filtered or unexported fields
}

MockFactory is a mock of Factory interface.

func NewMockFactory added in v1.17.3 

func NewMockFactory(ctrl *gomock.Controller) *MockFactory

NewMockFactory creates a new mock instance.

func (*MockFactory) CreateQueue added in v1.17.3 

func (m *MockFactory) CreateQueue(shard shard.Context, engine shard.Engine, cache workflow.Cache) Queue

CreateQueue mocks base method.

func (*MockFactory) EXPECT added in v1.17.3 

func (m *MockFactory) EXPECT() *MockFactoryMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockFactory) Start added in v1.17.3 

func (m *MockFactory) Start()

Start mocks base method.

func (*MockFactory) Stop added in v1.17.3 

func (m *MockFactory) Stop()

Stop mocks base method.

type MockFactoryMockRecorder added in v1.17.3 

type MockFactoryMockRecorder struct {
	// contains filtered or unexported fields
}

MockFactoryMockRecorder is the mock recorder for MockFactory.

func (*MockFactoryMockRecorder) CreateQueue added in v1.17.3 

func (mr *MockFactoryMockRecorder) CreateQueue(shard, engine, cache interface{}) *gomock.Call

CreateQueue indicates an expected call of CreateQueue.

func (*MockFactoryMockRecorder) Start added in v1.17.3 

func (mr *MockFactoryMockRecorder) Start() *gomock.Call

Start indicates an expected call of Start.

func (*MockFactoryMockRecorder) Stop added in v1.17.3 

func (mr *MockFactoryMockRecorder) Stop() *gomock.Call

Stop indicates an expected call of Stop.

type MockQueue added in v1.17.3 

type MockQueue struct {
	// contains filtered or unexported fields
}

MockQueue is a mock of Queue interface.

func NewMockQueue added in v1.17.3 

func NewMockQueue(ctrl *gomock.Controller) *MockQueue

NewMockQueue creates a new mock instance.

func (*MockQueue) Category added in v1.17.3 

func (m *MockQueue) Category() tasks.Category

Category mocks base method.

func (*MockQueue) EXPECT added in v1.17.3 

func (m *MockQueue) EXPECT() *MockQueueMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockQueue) FailoverNamespace added in v1.17.3 

func (m *MockQueue) FailoverNamespace(namespaceIDs map[string]struct{})

FailoverNamespace mocks base method.

func (*MockQueue) LockTaskProcessing added in v1.17.3 

func (m *MockQueue) LockTaskProcessing()

LockTaskProcessing mocks base method.

func (*MockQueue) NotifyNewTasks added in v1.17.3 

func (m *MockQueue) NotifyNewTasks(clusterName string, tasks []tasks.Task)

NotifyNewTasks mocks base method.

func (*MockQueue) Start added in v1.17.3 

func (m *MockQueue) Start()

Start mocks base method.

func (*MockQueue) Stop added in v1.17.3 

func (m *MockQueue) Stop()

Stop mocks base method.

func (*MockQueue) UnlockTaskProcessing added in v1.17.3 

func (m *MockQueue) UnlockTaskProcessing()

UnlockTaskProcessing mocks base method.

type MockQueueMockRecorder added in v1.17.3 

type MockQueueMockRecorder struct {
	// contains filtered or unexported fields
}

MockQueueMockRecorder is the mock recorder for MockQueue.

func (*MockQueueMockRecorder) Category added in v1.17.3 

func (mr *MockQueueMockRecorder) Category() *gomock.Call

Category indicates an expected call of Category.

func (*MockQueueMockRecorder) FailoverNamespace added in v1.17.3 

func (mr *MockQueueMockRecorder) FailoverNamespace(namespaceIDs interface{}) *gomock.Call

FailoverNamespace indicates an expected call of FailoverNamespace.

func (*MockQueueMockRecorder) LockTaskProcessing added in v1.17.3 

func (mr *MockQueueMockRecorder) LockTaskProcessing() *gomock.Call

LockTaskProcessing indicates an expected call of LockTaskProcessing.

func (*MockQueueMockRecorder) NotifyNewTasks added in v1.17.3 

func (mr *MockQueueMockRecorder) NotifyNewTasks(clusterName, tasks interface{}) *gomock.Call

NotifyNewTasks indicates an expected call of NotifyNewTasks.

func (*MockQueueMockRecorder) Start added in v1.17.3 

func (mr *MockQueueMockRecorder) Start() *gomock.Call

Start indicates an expected call of Start.

func (*MockQueueMockRecorder) Stop added in v1.17.3 

func (mr *MockQueueMockRecorder) Stop() *gomock.Call

Stop indicates an expected call of Stop.

func (*MockQueueMockRecorder) UnlockTaskProcessing added in v1.17.3 

func (mr *MockQueueMockRecorder) UnlockTaskProcessing() *gomock.Call

UnlockTaskProcessing indicates an expected call of UnlockTaskProcessing.

type MockRescheduler added in v1.17.0 

type MockRescheduler struct {
	// contains filtered or unexported fields
}

MockRescheduler is a mock of Rescheduler interface.

func NewMockRescheduler added in v1.17.0 

func NewMockRescheduler(ctrl *gomock.Controller) *MockRescheduler

NewMockRescheduler creates a new mock instance.

func (*MockRescheduler) Add added in v1.17.0 

func (m *MockRescheduler) Add(task Executable, rescheduleTime time.Time)

Add mocks base method.

func (*MockRescheduler) EXPECT added in v1.17.0 

func (m *MockRescheduler) EXPECT() *MockReschedulerMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockRescheduler) Len added in v1.17.0 

func (m *MockRescheduler) Len() int

Len mocks base method.

func (*MockRescheduler) Start added in v1.17.3 

func (m *MockRescheduler) Start()

Start mocks base method.

func (*MockRescheduler) Stop added in v1.17.3 

func (m *MockRescheduler) Stop()

Stop mocks base method.

type MockReschedulerMockRecorder added in v1.17.0 

type MockReschedulerMockRecorder struct {
	// contains filtered or unexported fields
}

MockReschedulerMockRecorder is the mock recorder for MockRescheduler.

func (*MockReschedulerMockRecorder) Add added in v1.17.0 

func (mr *MockReschedulerMockRecorder) Add(task, rescheduleTime interface{}) *gomock.Call

Add indicates an expected call of Add.

func (*MockReschedulerMockRecorder) Len added in v1.17.0 

func (mr *MockReschedulerMockRecorder) Len() *gomock.Call

Len indicates an expected call of Len.

func (*MockReschedulerMockRecorder) Start added in v1.17.3 

func (mr *MockReschedulerMockRecorder) Start() *gomock.Call

Start indicates an expected call of Start.

func (*MockReschedulerMockRecorder) Stop added in v1.17.3 

func (mr *MockReschedulerMockRecorder) Stop() *gomock.Call

Stop indicates an expected call of Stop.

type MockScheduler added in v1.17.0 

type MockScheduler struct {
	// contains filtered or unexported fields
}

MockScheduler is a mock of Scheduler interface.

func NewMockScheduler added in v1.17.0 

func NewMockScheduler(ctrl *gomock.Controller) *MockScheduler

NewMockScheduler creates a new mock instance.

func (*MockScheduler) EXPECT added in v1.17.0 

func (m *MockScheduler) EXPECT() *MockSchedulerMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockScheduler) Start added in v1.17.0 

func (m *MockScheduler) Start()

Start mocks base method.

func (*MockScheduler) Stop added in v1.17.0 

func (m *MockScheduler) Stop()

Stop mocks base method.

func (*MockScheduler) Submit added in v1.17.0 

func (m *MockScheduler) Submit(arg0 Executable) error

Submit mocks base method.

func (*MockScheduler) TrySubmit added in v1.17.0 

func (m *MockScheduler) TrySubmit(arg0 Executable) (bool, error)

TrySubmit mocks base method.

type MockSchedulerMockRecorder added in v1.17.0 

type MockSchedulerMockRecorder struct {
	// contains filtered or unexported fields
}

MockSchedulerMockRecorder is the mock recorder for MockScheduler.

func (*MockSchedulerMockRecorder) Start added in v1.17.0 

func (mr *MockSchedulerMockRecorder) Start() *gomock.Call

Start indicates an expected call of Start.

func (*MockSchedulerMockRecorder) Stop added in v1.17.0 

func (mr *MockSchedulerMockRecorder) Stop() *gomock.Call

Stop indicates an expected call of Stop.

func (*MockSchedulerMockRecorder) Submit added in v1.17.0 

func (mr *MockSchedulerMockRecorder) Submit(arg0 interface{}) *gomock.Call

Submit indicates an expected call of Submit.

func (*MockSchedulerMockRecorder) TrySubmit added in v1.17.0 

func (mr *MockSchedulerMockRecorder) TrySubmit(arg0 interface{}) *gomock.Call

TrySubmit indicates an expected call of TrySubmit.

type Options added in v1.17.3 

type Options struct {
	ReaderOptions

	MaxPollInterval                     dynamicconfig.DurationPropertyFn
	MaxPollIntervalJitterCoefficient    dynamicconfig.FloatPropertyFn
	CheckpointInterval                  dynamicconfig.DurationPropertyFn
	CheckpointIntervalJitterCoefficient dynamicconfig.FloatPropertyFn
	TaskMaxRetryCount                   dynamicconfig.IntPropertyFn
}

type PaginationFnProvider added in v1.17.3 

type PaginationFnProvider func(Range) collection.PaginationFn[tasks.Task]

type PriorityAssigner added in v1.17.0 

type PriorityAssigner interface {
	Assign(Executable) error
}

PriorityAssigner assigns priority to task executables

func NewNoopPriorityAssigner added in v1.17.0 

func NewNoopPriorityAssigner() PriorityAssigner

func NewPriorityAssigner added in v1.17.0 

func NewPriorityAssigner(
	currentClusterName string,
	namespaceRegistry namespace.Registry,
	options PriorityAssignerOptions,
	metricsProvider metrics.MetricsHandler,
) PriorityAssigner

type PriorityAssignerOptions added in v1.17.0 

type PriorityAssignerOptions struct {
	HighPriorityRPS       dynamicconfig.IntPropertyFnWithNamespaceFilter
	CriticalRetryAttempts dynamicconfig.IntPropertyFn
}

type Queue added in v1.17.3 

type Queue interface {
	common.Daemon
	Category() tasks.Category
	NotifyNewTasks(clusterName string, tasks []tasks.Task)
	FailoverNamespace(namespaceIDs map[string]struct{})
	LockTaskProcessing()
	UnlockTaskProcessing()
}

type QueueType added in v1.17.0 

type QueueType int

TODO: remove QueueType after merging active and standby transfer/timer queue. Use tasks.Category instead Currently need queue active/standby information for assigning priority 

const (
	QueueTypeUnknown QueueType = iota
	// QueueTypeTransfer is used by single cursor transfer queue, which
	// processes both active and standby task
	QueueTypeTransfer
	QueueTypeActiveTransfer
	QueueTypeStandbyTransfer
	// QueueTypeTimer is used by single cursor timer queue, which
	// processes both active and standby task
	QueueTypeTimer
	QueueTypeActiveTimer
	QueueTypeStandbyTimer
	QueueTypeVisibility
)

type Range added in v1.17.0 

type Range struct {
	InclusiveMin tasks.Key
	ExclusiveMax tasks.Key
}

func FromPersistenceRange added in v1.17.3 

func FromPersistenceRange(
	r *persistencespb.QueueSliceRange,
) Range

func NewRandomOrderedRangesInRange added in v1.17.3 

func NewRandomOrderedRangesInRange(
	r Range,
	numRanges int,
) []Range

func NewRandomRange added in v1.17.0 

func NewRandomRange() Range

func NewRange added in v1.17.0 

func NewRange(
	inclusiveMin tasks.Key,
	exclusiveMax tasks.Key,
) Range

func (*Range) CanMerge added in v1.17.0 

func (r *Range) CanMerge(
	input Range,
) bool

func (*Range) CanSplit added in v1.17.0 

func (r *Range) CanSplit(
	key tasks.Key,
) bool

func (*Range) ContainsKey added in v1.17.0 

func (r *Range) ContainsKey(
	key tasks.Key,
) bool

func (*Range) ContainsRange added in v1.17.0 

func (r *Range) ContainsRange(
	input Range,
) bool

func (*Range) Equals added in v1.17.3 

func (r *Range) Equals(
	input Range,
) bool

func (*Range) IsEmpty added in v1.17.0 

func (r *Range) IsEmpty() bool

func (*Range) Merge added in v1.17.0 

func (r *Range) Merge(
	input Range,
) Range

func (*Range) Split added in v1.17.0 

func (r *Range) Split(
	key tasks.Key,
) (left Range, right Range)

type Reader added in v1.17.3 

type Reader interface {
	common.Daemon

	Scopes() []Scope

	WalkSlices(SliceIterator)
	SplitSlices(SliceSplitter)
	MergeSlices(...Slice)
	ClearSlices(SlicePredicate)
	ShrinkSlices()

	Pause(time.Duration)
}

type ReaderGroup added in v1.17.3 

type ReaderGroup struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewReaderGroup added in v1.17.3 

func NewReaderGroup(
	initializer ReaderInitializer,
) *ReaderGroup

func (*ReaderGroup) NewReader added in v1.17.3 

func (g *ReaderGroup) NewReader(readerID int32, slices ...Slice) Reader

func (*ReaderGroup) ReaderByID added in v1.17.3 

func (g *ReaderGroup) ReaderByID(readerID int32) (Reader, bool)

func (*ReaderGroup) Readers added in v1.17.3 

func (g *ReaderGroup) Readers() map[int32]Reader

func (*ReaderGroup) Start added in v1.17.3 

func (g *ReaderGroup) Start()

func (*ReaderGroup) Stop added in v1.17.3 

func (g *ReaderGroup) Stop()

type ReaderImpl added in v1.17.3 

type ReaderImpl struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewReader added in v1.17.3 

func NewReader(
	slices []Slice,
	options *ReaderOptions,
	scheduler Scheduler,
	rescheduler Rescheduler,
	timeSource clock.TimeSource,
	ratelimiter quotas.RateLimiter,
	logger log.Logger,
	metricsHandler metrics.MetricsHandler,
) *ReaderImpl

func (*ReaderImpl) ClearSlices added in v1.17.3 

func (r *ReaderImpl) ClearSlices(selector SlicePredicate)

func (*ReaderImpl) MergeSlices added in v1.17.3 

func (r *ReaderImpl) MergeSlices(incomingSlices ...Slice)

func (*ReaderImpl) Pause added in v1.17.3 

func (r *ReaderImpl) Pause(duration time.Duration)

func (*ReaderImpl) Scopes added in v1.17.3 

func (r *ReaderImpl) Scopes() []Scope

func (*ReaderImpl) ShrinkSlices added in v1.17.3 

func (r *ReaderImpl) ShrinkSlices()

func (*ReaderImpl) SplitSlices added in v1.17.3 

func (r *ReaderImpl) SplitSlices(splitter SliceSplitter)

func (*ReaderImpl) Start added in v1.17.3 

func (r *ReaderImpl) Start()

func (*ReaderImpl) Stop added in v1.17.3 

func (r *ReaderImpl) Stop()

func (*ReaderImpl) WalkSlices added in v1.17.3 

func (r *ReaderImpl) WalkSlices(iterator SliceIterator)

type ReaderInitializer added in v1.17.3 

type ReaderInitializer func(readerID int32, slices []Slice) Reader

type ReaderOptions added in v1.17.3 

type ReaderOptions struct {
	BatchSize            dynamicconfig.IntPropertyFn
	MaxPendingTasksCount dynamicconfig.IntPropertyFn
	PollBackoffInterval  dynamicconfig.DurationPropertyFn
}

type Rescheduler added in v1.17.0 

type Rescheduler interface {
	common.Daemon

	// Add task executable to the rescheduler.
	Add(task Executable, rescheduleTime time.Time)

	// Len returns the total number of task executables waiting to be rescheduled.
	Len() int
}

Rescheduler buffers task executables that are failed to process and resubmit them to the task scheduler when the Reschedule method is called.

type Scheduler added in v1.17.0 

type Scheduler interface {
	common.Daemon

	Submit(Executable) error
	TrySubmit(Executable) (bool, error)
}

Scheduler is the component for scheduling and processing task executables. Ack(), Nack() or Reschedule() will always be called on all executables that have been successfully submited. Reschedule() will only be called after the Scheduler has been stopped

type SchedulerOptions added in v1.17.0 

type SchedulerOptions struct {
	tasks.ParallelProcessorOptions
	tasks.InterleavedWeightedRoundRobinSchedulerOptions
}

type Scope added in v1.17.0 

type Scope struct {
	Range     Range
	Predicate tasks.Predicate
}

func FromPersistenceScope added in v1.17.3 

func FromPersistenceScope(
	scope *persistencespb.QueueSliceScope,
) Scope

func NewRandomScopes added in v1.17.3 

func NewRandomScopes(
	numScopes int,
) []Scope

func NewScope added in v1.17.0 

func NewScope(
	r Range,
	predicate tasks.Predicate,
) Scope

func (*Scope) CanMergeByPredicate added in v1.17.0 

func (s *Scope) CanMergeByPredicate(
	incomingScope Scope,
) bool

func (*Scope) CanMergeByRange added in v1.17.0 

func (s *Scope) CanMergeByRange(
	incomingScope Scope,
) bool

func (*Scope) CanSplitByRange added in v1.17.0 

func (s *Scope) CanSplitByRange(
	key tasks.Key,
) bool

func (*Scope) Contains added in v1.17.0 

func (s *Scope) Contains(task tasks.Task) bool

func (*Scope) Equals added in v1.17.3 

func (s *Scope) Equals(scope Scope) bool

func (*Scope) IsEmpty added in v1.17.3 

func (s *Scope) IsEmpty() bool

func (*Scope) MergeByPredicate added in v1.17.0 

func (s *Scope) MergeByPredicate(
	incomingScope Scope,
) Scope

func (*Scope) MergeByRange added in v1.17.0 

func (s *Scope) MergeByRange(
	incomingScope Scope,
) Scope

func (*Scope) SplitByPredicate added in v1.17.0 

func (s *Scope) SplitByPredicate(
	predicate tasks.Predicate,
) (pass Scope, fail Scope)

func (*Scope) SplitByRange added in v1.17.0 

func (s *Scope) SplitByRange(
	key tasks.Key,
) (left Scope, right Scope)

type Slice added in v1.17.3 

type Slice interface {
	Scope() Scope
	CanSplitByRange(tasks.Key) bool
	SplitByRange(tasks.Key) (left Slice, right Slice)
	SplitByPredicate(tasks.Predicate) (pass Slice, fail Slice)
	CanMergeWithSlice(Slice) bool
	MergeWithSlice(Slice) []Slice
	ShrinkRange()
	SelectTasks(int) ([]Executable, error)
	MoreTasks() bool
	Clear()
}

Slice manages the loading and status tracking of all tasks within its Scope. It also provides methods for splitting or merging with another slice either by range or by predicate.

type SliceImpl added in v1.17.3 

type SliceImpl struct {
	// contains filtered or unexported fields
}

func NewSlice added in v1.17.3 

func NewSlice(
	paginationFnProvider PaginationFnProvider,
	executableInitializer ExecutableInitializer,
	scope Scope,
) *SliceImpl

func (*SliceImpl) CanMergeWithSlice added in v1.17.3 

func (s *SliceImpl) CanMergeWithSlice(slice Slice) bool

func (*SliceImpl) CanSplitByRange added in v1.17.3 

func (s *SliceImpl) CanSplitByRange(key tasks.Key) bool

func (*SliceImpl) Clear added in v1.17.3 

func (s *SliceImpl) Clear()

func (*SliceImpl) MergeWithSlice added in v1.17.3 

func (s *SliceImpl) MergeWithSlice(slice Slice) []Slice

func (*SliceImpl) MoreTasks added in v1.17.3 

func (s *SliceImpl) MoreTasks() bool

func (*SliceImpl) Scope added in v1.17.3 

func (s *SliceImpl) Scope() Scope

func (*SliceImpl) SelectTasks added in v1.17.3 

func (s *SliceImpl) SelectTasks(batchSize int) ([]Executable, error)

func (*SliceImpl) ShrinkRange added in v1.17.3 

func (s *SliceImpl) ShrinkRange()

func (*SliceImpl) SplitByPredicate added in v1.17.3 

func (s *SliceImpl) SplitByPredicate(predicate tasks.Predicate) (pass Slice, fail Slice)

func (*SliceImpl) SplitByRange added in v1.17.3 

func (s *SliceImpl) SplitByRange(key tasks.Key) (left Slice, right Slice)

type SliceIterator added in v1.17.3 

type SliceIterator func(s Slice)

type SlicePredicate added in v1.17.3 

type SlicePredicate func(s Slice) bool

type SliceSplitter added in v1.17.3 

type SliceSplitter func(s Slice) (remaining []Slice, split bool)

type TaskFilter added in v1.17.0 

type TaskFilter func(task tasks.Task) bool

TaskFilter determines if the given task should be executed TODO: remove after merging active/standby queue processor task should always be executed as active or verified as standby

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.