worker

Documentation

Index

• Variables
• func Cleanqueue()
• func CloseWorkerPools()
• func CreateCronWorker()
• func Dispatch(name string, fn func(), queue string) error
• func DispatchCron(cronStr string, name string, queue string, fn func()) error
• func DispatchEvery(interval time.Duration, name string, queue string, fn func()) error
• func GetQueues() map[string]dispatcherQueue
• func GetSchedules() map[string]JobSchedule
• func InitWorkerPools(workerindexer int, workerparse int, workersearch int, workerfiles int, ...)
• func Stack() string
• func StartCronWorker()
• func StopCronWorker()
• type Job
• type JobSchedule

Variables

var WorkerPoolFiles *pond.WorkerPool

workerPoolFiles is a WorkerPool for executing file tasks

var WorkerPoolIndexer *pond.WorkerPool

workerPoolIndexer is a WorkerPool for executing indexer tasks

var WorkerPoolMetadata *pond.WorkerPool

workerPoolMetadata is a WorkerPool for executing metadata tasks

var WorkerPoolParse *pond.WorkerPool

workerPoolParse is a WorkerPool for executing parse tasks

var WorkerPoolSearch *pond.WorkerPool

workerPoolSearch is a WorkerPool for executing search tasks

Functions

func Cleanqueue

func Cleanqueue()

Cleanqueue clears the global queue set if there are no running or waiting workers across all pools.

func CloseWorkerPools

func CloseWorkerPools()

CloseWorkerPools stops all worker pools and waits for workers to finish current jobs before returning. Waits up to 2 minutes per pool before timing out.

func CreateCronWorker

func CreateCronWorker()

CreateCronWorker initializes the cron workers for data, feeds, and search. It configures each cron worker with the application logger, sets the timezone, adds error recovery and duplicate job prevention middleware, and enables running jobs at a per-second interval.

func Dispatch

func Dispatch(name string, fn func(), queue string) error

Dispatch adds a new job with the given name, function, and queue to the worker pool. It generates a new UUID to associate with the job.

func DispatchCron

func DispatchCron(cronStr string, name string, queue string, fn func()) error

DispatchCron schedules a cron job to run the given function fn at the specified cron schedule cronStr. It adds the job to the worker queue specified by queue and gives it the name name. It returns any error from setting up the cron job.

func DispatchEvery

func DispatchEvery(interval time.Duration, name string, queue string, fn func()) error

DispatchEvery dispatches a job to run on a regular time interval. It takes in the interval duration, job name, queue, and function to run. It returns any error from setting up the ticker.

func GetQueues

func GetQueues() map[string]dispatcherQueue

GetQueues returns a map of all currently configured queues, keyed by the queue name.

func GetSchedules

func GetSchedules() map[string]JobSchedule

GetSchedules returns a map of all currently configured schedules, keyed by the job name.

func InitWorkerPools

func InitWorkerPools(workerindexer int, workerparse int, workersearch int, workerfiles int, workermeta int)

InitWorkerPools initializes the worker pools for indexing, parsing, searching, downloading files, and updating metadata. It takes in the desired number of workers for each pool and defaults them to 1 if 0 is passed in. It configures the pools with balanced strategy and error handling function.

func Stack

func Stack() string

func StartCronWorker

func StartCronWorker()

StartCronWorker starts all cron workers.

func StopCronWorker

func StopCronWorker()

StopCronWorker stops all cron workers.

Types

type Job

type Job struct {
	// Queue is the name of the queue this job belongs to
	Queue string
	// ID is a unique identifier for this job
	ID string
	// Added is the time this job was added to the queue
	Added time.Time
	// Started is the time this job was started by a worker
	Started time.Time
	// Name is a descriptive name for this job
	Name string
	// SchedulerID is the ID of the scheduler that added this job
	SchedulerID string
	// Run is the function to execute for this job
	Run func() `json:"-"`
	// CronJob is the cron job instance if this is a recurring cron job
	CronJob cron.Job `json:"-"`
}

Job represents a job to be run by a worker pool

type JobSchedule

type JobSchedule struct {
	// JobName is the name of the job
	JobName string
	// JobID is the unique ID of the job
	JobID string
	// ID is the unique ID for this schedule
	ID string
	// ScheduleTyp is the type of schedule (cron, interval, etc)
	ScheduleTyp string
	// ScheduleString is the schedule string (cron expression, interval, etc)
	ScheduleString string
	// Interval is the interval duration if schedule type is interval
	Interval time.Duration
	// CronSchedule is the parsed cron.Schedule if type is cron
	CronSchedule cron.Schedule
	// CronID is the cron scheduler ID if scheduled as cron job
	CronID cron.EntryID

	// LastRun is the last time this job ran
	LastRun time.Time
	// NextRun is the next scheduled run time
	NextRun time.Time
	// IsRunning indicates if the job is currently running
	IsRunning bool
}

JobSchedule represents a scheduled job