taskmanager

Documentation

Overview

Package taskmanager is the task/job management service layer for concurrent and asynchronous tasks with cron scheduling.

Index

• func DefaultTaskQConfig(name string, opts ...TasqOps) *taskq.QueueOptions
• type CronJob
• type CronJobHandler
• type CronJobs
• type Factory
  • func (f Factory) IsEmpty() bool
  • func (f Factory) String() string
• type TaskEngine
  • func NewTaskManager(ctx context.Context, opts ...TaskManagerOptions) (TaskEngine, error)
• type TaskManager
  • func (tm *TaskManager) Close(ctx context.Context) error
  • func (tm *TaskManager) CronJobsInit(cronJobsMap CronJobs) (err error)
  • func (tm *TaskManager) Factory() Factory
  • func (tm *TaskManager) GetTxnCtx(ctx context.Context) context.Context
  • func (tm *TaskManager) IsNewRelicEnabled() bool
  • func (c *TaskManager) RegisterTask(name string, handler interface{}) (err error)
  • func (tm *TaskManager) ResetCron()
  • func (c *TaskManager) RunTask(ctx context.Context, options *TaskRunOptions) error
  • func (tm *TaskManager) Tasks() map[string]*taskq.Task
• type TaskManagerOptions
  • func WithLogger(customLogger *zerolog.Logger) TaskManagerOptions
  • func WithNewRelic() TaskManagerOptions
  • func WithTaskqConfig(config *taskq.QueueOptions) TaskManagerOptions
• type TaskRunOptions
• type TasqOps
  • func WithRedis(addr string) TasqOps

Functions

func DefaultTaskQConfig

func DefaultTaskQConfig(name string, opts ...TasqOps) *taskq.QueueOptions

DefaultTaskQConfig will return a QueueOptions with specified name and functional options applied

Types

type CronJob

type CronJob struct {
	Handler CronJobHandler
	Period  time.Duration
}

CronJob definition, params reduced to the minimum, all required

type CronJobHandler

type CronJobHandler func(ctx context.Context) error

CronJobHandler is the handler for a cron job

type CronJobs

type CronJobs map[string]CronJob

CronJobs as a map prevents duplicate jobs with the same name

type Factory

type Factory string

Factory is the different types of task factories that are supported

const (
	FactoryEmpty  Factory = "empty"
	FactoryMemory Factory = "memory"
	FactoryRedis  Factory = "redis"
)

Supported factories

func (Factory) IsEmpty

func (f Factory) IsEmpty() bool

IsEmpty will return true if the factory is not set

func (Factory) String

func (f Factory) String() string

String is the string version of factory

type TaskEngine

type TaskEngine interface {
	RegisterTask(name string, handler interface{}) error
	ResetCron()
	RunTask(ctx context.Context, options *TaskRunOptions) error
	Tasks() map[string]*taskq.Task
	CronJobsInit(cronJobsMap CronJobs) error
	Close(ctx context.Context) error
	Factory() Factory
	GetTxnCtx(ctx context.Context) context.Context
	IsNewRelicEnabled() bool
}

TaskEngine is the taskmanager client interface

func NewTaskManager

func NewTaskManager(ctx context.Context, opts ...TaskManagerOptions) (TaskEngine, error)

NewTaskManager creates a new client for all TaskManager functionality If no options are given, it will use local memory for the queue. ctx may contain a NewRelic txn (or one will be created)

type TaskManager

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

TaskManager implements the TaskEngine interface

func (*TaskManager) Close

func (tm *TaskManager) Close(ctx context.Context) error

Close the client and any open connections

func (*TaskManager) CronJobsInit

func (tm *TaskManager) CronJobsInit(cronJobsMap CronJobs) (err error)

CronJobsInit registers and runs the cron jobs

func (*TaskManager) Factory

func (tm *TaskManager) Factory() Factory

Factory will return the factory that is set

func (*TaskManager) GetTxnCtx

func (tm *TaskManager) GetTxnCtx(ctx context.Context) context.Context

GetTxnCtx will check for an existing transaction

func (*TaskManager) IsNewRelicEnabled

func (tm *TaskManager) IsNewRelicEnabled() bool

IsNewRelicEnabled will return if new relic is enabled

func (*TaskManager) RegisterTask

func (c *TaskManager) RegisterTask(name string, handler interface{}) (err error)

RegisterTask will register a new task to handle asynchronously

func (*TaskManager) ResetCron

func (tm *TaskManager) ResetCron()

ResetCron will reset the cron scheduler and all loaded tasks

func (*TaskManager) RunTask

func (c *TaskManager) RunTask(ctx context.Context, options *TaskRunOptions) error

RunTask will run a task using TaskQ

func (*TaskManager) Tasks

func (tm *TaskManager) Tasks() map[string]*taskq.Task

Tasks will return the list of tasks

type TaskManagerOptions

type TaskManagerOptions func(c *options)

TaskManagerOptions allow functional options to be supplied

func WithLogger

func WithLogger(customLogger *zerolog.Logger) TaskManagerOptions

WithLogger will set the custom logger interface

func WithNewRelic

func WithNewRelic() TaskManagerOptions

WithNewRelic will enable the NewRelic wrapper

func WithTaskqConfig

func WithTaskqConfig(config *taskq.QueueOptions) TaskManagerOptions

WithTaskqConfig will set the taskq custom config

type TaskRunOptions

type TaskRunOptions struct {
	Arguments      []interface{} // Arguments for the task
	RunEveryPeriod time.Duration // Cron job!
	TaskName       string        // Name of the task
}

TaskRunOptions are the options for running a task

type TasqOps

type TasqOps func(*taskq.QueueOptions)

TasqOps allow functional options to be supplied

func WithRedis

func WithRedis(addr string) TasqOps

WithRedis will set the redis client for the TaskQ engine Note: Because we use redis/v8, we need to use Redis lower than 7.2.0