asyncjobs

package module

Go to main page

Versions in this module

v0

v0.1.2

Jan 11, 2023

v0.1.1

Dec 26, 2022

v0.1.0

Dec 18, 2022

Changes in this version

+ var ErrTaskDependenciesFailed = fmt.Errorf("task dependencies failed")

+ var ErrUnknownDiscardPolicy = fmt.Errorf("unknown discard policy")

+ func RequestReplySubjectForTaskType(taskType string) string

type ClientOpt

+ func DiscardTaskStatesByName(states ...string) ClientOpt

type Storage

+ NakBlockedItem func(ctx context.Context, item *ProcessItem) error

type Task

+ Dependencies []string

+ DependencyResults map[string]*TaskResult

+ LoadDependencies bool

+ func (t *Task) HasDependencies() bool

type TaskOpt

+ func TaskDependsOn(tasks ...*Task) TaskOpt

+ func TaskDependsOnIDs(ids ...string) TaskOpt

+ func TaskRequiresDependencyResults() TaskOpt

type TaskState

+ const TaskStateBlocked

+ const TaskStateUnreachable

v0.0.7

Mar 4, 2022

Changes in this version

+ var ErrExternalCommandFailed = fmt.Errorf("execution failed")

+ var ErrExternalCommandNotFound = fmt.Errorf("command not found")

type Mux

+ func (m *Mux) ExternalProcess(taskType string, command string) error

v0.0.6

Feb 26, 2022

v0.0.5

Feb 18, 2022

Changes in this version

+ const ConfigBucketName

+ const LeaderElectedEventSubjectPattern

+ const LeaderElectedEventSubjectWildcard

+ const LeaderElectedEventType

+ const LeaderElectionBucketName

+ const ShortedScheduledDeadline

+ var ErrScheduleInvalid = errors.New("invalid cron schedule")

+ var ErrScheduleIsRequired = errors.New("schedule is required")

+ var ErrScheduleNameInvalid = errors.New("name is invalid")

+ var ErrScheduleNameIsRequired = errors.New("name is required")

+ var ErrScheduledTaskAlreadyExist = errors.New("scheduled task already exist")

+ var ErrScheduledTaskInvalid = errors.New("invalid scheduled task")

+ var ErrScheduledTaskNotFound = errors.New("scheduled task not found")

+ var ErrScheduledTaskShortDeadline = errors.New("deadline too short")

+ var ErrStorageNotReady = fmt.Errorf("storage not ready")

+ var ErrTaskExceedsMaxTries = fmt.Errorf("exceeded maximum tries")

+ var ErrTaskTypeInvalid = fmt.Errorf("task type is invalid")

+ var ErrTaskTypeRequired = fmt.Errorf("task type is required")

+ var ErrUnknownRetryPolicy = fmt.Errorf("unknown retry policy")

+ func IsRetryPolicyKnown(name string) bool

+ func IsValidName(name string) bool

+ func RetryPolicyNames() []string

type Client

+ func (c *Client) LoadScheduledTaskByName(name string) (*ScheduledTask, error)

+ func (c *Client) NewScheduledTask(name string, schedule string, queue string, task *Task) error

+ func (c *Client) RemoveScheduledTask(name string) error

+ func (c *Client) ScheduledTasksStorage() ScheduledTaskStorage

type ClientOpt

+ func RetryBackoffPolicyName(name string) ClientOpt

+ type LeaderElectedEvent struct

+ Component string

+ Name string

+ func NewLeaderElectedEvent(name string, component string) (*LeaderElectedEvent, error)

type RetryPolicyProvider

+ func RetryPolicyLookup(name string) (RetryPolicyProvider, error)

+ type ScheduleWatchEntry struct

+ Delete bool

+ Name string

+ Task *ScheduledTask

+ type ScheduledTask struct

+ CreatedAt time.Time

+ Deadline time.Duration

+ MaxTries int

+ Name string

+ Payload []byte

+ Queue string

+ Schedule string

+ TaskType string

+ type ScheduledTaskStorage interface

+ DeleteScheduledTaskByName func(name string) error

+ ElectionStorage func() (nats.KeyValue, error)

+ EnqueueTask func(ctx context.Context, queue *Queue, task *Task) error

+ LoadScheduledTaskByName func(name string) (*ScheduledTask, error)

+ PublishLeaderElectedEvent func(ctx context.Context, name string, component string) error

+ SaveScheduledTask func(st *ScheduledTask, update bool) error

+ ScheduledTasks func(ctx context.Context) ([]*ScheduledTask, error)

+ ScheduledTasksWatch func(ctx context.Context) (chan *ScheduleWatchEntry, error)

type Storage

+ DeleteScheduledTaskByName func(name string) error

+ DeleteTaskByID func(id string) error

+ LoadScheduledTaskByName func(name string) (*ScheduledTask, error)

+ PrepareConfigurationStore func(memory bool, replicas int) error

+ SaveScheduledTask func(st *ScheduledTask, update bool) error

+ ScheduledTasks func(ctx context.Context) ([]*ScheduledTask, error)

+ ScheduledTasksWatch func(ctx context.Context) (chan *ScheduleWatchEntry, error)

type StorageAdmin

+ ConfigurationInfo func() (*nats.KeyValueBucketStatus, error)

+ ElectionStorage func() (nats.KeyValue, error)

+ PrepareConfigurationStore func(memory bool, replicas int) error

type Task

+ MaxTries int

type TaskOpt

+ func TaskMaxTries(tries int) TaskOpt

+ type TaskScheduler struct

+ func NewTaskScheduler(name string, c *Client) (*TaskScheduler, error)

+ func (s *TaskScheduler) Count() int

+ func (s *TaskScheduler) Run(ctx context.Context, wg *sync.WaitGroup) error

+ func (s *TaskScheduler) Stop()

v0.0.4

Feb 9, 2022

Changes in this version

+ const RequestReplyContentTypeHeader

+ const RequestReplyDeadlineHeader

+ const RequestReplyError

+ const RequestReplyTaskHandlerPattern

+ const RequestReplyTaskType

+ const RequestReplyTerminateError

+ var ErrRequestReplyFailed = fmt.Errorf("request-reply callout failed")

+ var ErrRequestReplyNoDeadline = fmt.Errorf("request-reply requires deadline context")

+ var ErrRequestReplyShortDeadline = fmt.Errorf("request-reply deadline too short")

type Mux

+ func (m *Mux) RequestReply(taskType string, client *Client) error

v0.0.3

Feb 8, 2022

Changes in this version

+ var ErrContextWithoutDeadline = fmt.Errorf("non deadline context given")

+ var ErrDuplicateHandlerForTaskType = fmt.Errorf("duplicate handler for task type")

+ var ErrDuplicateItem = fmt.Errorf("duplicate work queue item")

+ var ErrInvalidHeaders = fmt.Errorf("coult not decode headers")

+ var ErrInvalidQueueState = fmt.Errorf("invalid queue storage state")

+ var ErrInvalidStorageItem = fmt.Errorf("invalid storage item")

+ var ErrNoHandlerForTaskType = fmt.Errorf("no handler for task type")

+ var ErrNoMux = fmt.Errorf("mux is required")

+ var ErrNoNatsConn = fmt.Errorf("no NATS connection supplied")

+ var ErrNoTasks = fmt.Errorf("no tasks found")

+ var ErrQueueConsumerNotFound = errors.New("queue consumer not found")

+ var ErrQueueItemCorrupt = fmt.Errorf("corrupt queue item received")

+ var ErrQueueItemInvalid = fmt.Errorf("invalid queue item received")

+ var ErrQueueNameRequired = fmt.Errorf("queue name is required")

+ var ErrTaskAlreadyActive = fmt.Errorf("task already active")

+ var ErrTaskAlreadyInState = fmt.Errorf("%w, already in desired state", ErrTaskUpdateFailed)

+ var ErrTaskLoadFailed = fmt.Errorf("loading task failed")

+ var ErrTaskPastDeadline = fmt.Errorf("past deadline")

+ var ErrTaskTypeCannotEnqueue = fmt.Errorf("cannot enqueue a task in state")

+ var ErrTaskUpdateFailed = fmt.Errorf("failed updating task state")

+ var ErrUnknownEventType = fmt.Errorf("unknown event type")

v0.0.2

Feb 7, 2022

Changes in this version

+ const EventsSubjectWildcard

+ const TaskStateChangeEventSubjectPattern

+ const TaskStateChangeEventSubjectWildcard

+ const TaskStateChangeEventType

+ func ParseEventJSON(event []byte) (interface{}, string, error)

+ func RetrySleep(ctx context.Context, p RetryPolicyProvider, n int) error

+ type BaseEvent struct

+ EventID string

+ EventType string

+ TimeStamp int64

type Client

+ func (c *Client) RetryTaskByID(ctx context.Context, id string) error

type ClientOpt

+ func DiscardTaskStates(states ...TaskState) ClientOpt

+ func NoStorageInit() ClientOpt

+ type RetryPolicyProvider interface

+ Duration func(n int) time.Duration

type Storage

+ PublishTaskStateChangeEvent func(ctx context.Context, task *Task) error

+ RetryTaskByID func(ctx context.Context, queue *Queue, id string) error

type Task

+ func (t *Task) IsPastDeadline() bool

+ type TaskStateChangeEvent struct

+ Age time.Duration

+ LastErr string

+ Queue string

+ State TaskState

+ TaskID string

+ TaskType string

+ Tries int

+ func NewTaskStateChangeEvent(t *Task) (*TaskStateChangeEvent, error)

v0.0.1

Feb 2, 2022

Changes in this version

+ const DefaultJobRunTime

+ const DefaultMaxTries

+ const DefaultQueueMaxConcurrent

+ const TasksStreamName

+ const TasksStreamSubjectPattern

+ const TasksStreamSubjects

+ const WorkStreamNamePattern

+ const WorkStreamNamePrefix

+ const WorkStreamSubjectPattern

+ const WorkStreamSubjectWildcard

+ var ErrQueueNotFound = errors.New("queue not found")

+ var ErrTaskNotFound = errors.New("task not found")

+ var ErrTerminateTask = fmt.Errorf("terminate task")

+ var RetryDefault = RetryLinearTenMinutes

+ var RetryLinearOneHour = linearPolicy(20, 0.90, 10 * time.Minute, 60 * time.Minute)

+ var RetryLinearOneMinute = linearPolicy(20, 0.5, time.Second, time.Minute)

+ var RetryLinearTenMinutes = linearPolicy(20, 0.90, time.Minute, 10 * time.Minute)

+ type Client struct

+ func NewClient(opts ...ClientOpt) (*Client, error)

+ func (c *Client) EnqueueTask(ctx context.Context, task *Task) error

+ func (c *Client) LoadTaskByID(id string) (*Task, error)

+ func (c *Client) Run(ctx context.Context, router *Mux) error

+ func (c *Client) StorageAdmin() StorageAdmin

+ type ClientOpt func(opts *ClientOpts) error

+ func BindWorkQueue(queue string) ClientOpt

+ func ClientConcurrency(c int) ClientOpt

+ func CustomLogger(log Logger) ClientOpt

+ func MemoryStorage() ClientOpt

+ func NatsConn(nc *nats.Conn) ClientOpt

+ func NatsContext(c string, opts ...nats.Option) ClientOpt

+ func PrometheusListenPort(port int) ClientOpt

+ func RetryBackoffPolicy(p RetryPolicy) ClientOpt

+ func StoreReplicas(r uint) ClientOpt

+ func TaskRetention(r time.Duration) ClientOpt

+ func WorkQueue(queue *Queue) ClientOpt

+ type ClientOpts struct

+ type Handler interface

+ ProcessTask func(ctx context.Context, t *Task) (interface{}, error)

+ type HandlerFunc func(ctx context.Context, t *Task) (interface{}, error)

+ type ItemKind int

+ var TaskItem ItemKind = 0

+ type Logger interface

+ Debugf func(format string, v ...interface{})

+ Errorf func(format string, v ...interface{})

+ Infof func(format string, v ...interface{})

+ Warnf func(format string, v ...interface{})

+ type Mux struct

+ func NewTaskRouter() *Mux

+ func (m *Mux) HandleFunc(taskType string, h HandlerFunc) error

+ func (m *Mux) Handler(t *Task) HandlerFunc

+ type ProcessItem struct

+ JobID string

+ Kind ItemKind

+ type Queue struct

+ DiscardOld bool

+ MaxAge time.Duration

+ MaxConcurrent int

+ MaxEntries int

+ MaxRunTime time.Duration

+ MaxTries int

+ Name string

+ NoCreate bool

+ type QueueInfo struct

+ Consumer *api.ConsumerInfo

+ Name string

+ Stream *api.StreamInfo

+ Time time.Time

+ type RetryPolicy struct

+ Intervals []time.Duration

+ Jitter float64

+ func (b RetryPolicy) Duration(n int) time.Duration

+ func (b RetryPolicy) Sleep(ctx context.Context, n int) error

+ type Storage interface

+ AckItem func(ctx context.Context, item *ProcessItem) error

+ EnqueueTask func(ctx context.Context, queue *Queue, task *Task) error

+ LoadTaskByID func(id string) (*Task, error)

+ NakItem func(ctx context.Context, item *ProcessItem) error

+ PollQueue func(ctx context.Context, q *Queue) (*ProcessItem, error)

+ PrepareQueue func(q *Queue, replicas int, memory bool) error

+ PrepareTasks func(memory bool, replicas int, retention time.Duration) error

+ SaveTaskState func(ctx context.Context, task *Task) error

+ TerminateItem func(ctx context.Context, item *ProcessItem) error

+ type StorageAdmin interface

+ DeleteQueue func(name string) error

+ DeleteTaskByID func(id string) error

+ LoadTaskByID func(id string) (*Task, error)

+ PrepareQueue func(q *Queue, replicas int, memory bool) error

+ PrepareTasks func(memory bool, replicas int, retention time.Duration) error

+ PurgeQueue func(name string) error

+ QueueInfo func(name string) (*QueueInfo, error)

+ QueueNames func() ([]string, error)

+ Queues func() ([]*QueueInfo, error)

+ Tasks func(ctx context.Context, limit int32) (chan *Task, error)

+ TasksInfo func() (*TasksInfo, error)

+ TasksStore func() (*jsm.Manager, *jsm.Stream, error)

+ type Task struct

+ CreatedAt time.Time

+ Deadline *time.Time

+ ID string

+ LastErr string

+ LastTriedAt *time.Time

+ Payload []byte

+ Queue string

+ Result *TaskResult

+ State TaskState

+ Tries int

+ Type string

+ func NewTask(taskType string, payload interface{}, opts ...TaskOpt) (*Task, error)

+ type TaskOpt func(*Task) error

+ func TaskDeadline(deadline time.Time) TaskOpt

+ type TaskResult struct

+ CompletedAt time.Time

+ Payload interface{}

+ type TaskState string

+ const TaskStateActive

+ const TaskStateCompleted

+ const TaskStateExpired

+ const TaskStateNew

+ const TaskStateQueueError

+ const TaskStateRetry

+ const TaskStateTerminated

+ const TaskStateUnknown

+ type TasksInfo struct

+ Stream *api.StreamInfo

+ Time time.Time

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL