background_tasks

README

Summary

The background_tasks package is an internal package responsible for managing background jobs within DMS. It contains a scheduler that registers tasks and run them according to the schedule defined by the task definition.

Tasks

Task is a struct that defines a job. It includes the task's ID, Name, the function that is going to be run, the arguments for the function, the triggers that trigger the task to run, retry policy, etc.

Triggers

Trigger is an interface that defines IsReady and Reset methods. IsReady should return true if the task should be run and Reset resets the trigger until the next event happens. There are different implementations for the trigger interface.

• PeriodicTrigger: Defines a trigger based on a duration interval or a cron expression.
• EventTrigger: Defines a trigger that is set by a trigger channel.
• OneTimeTrigger: A trigger that is only triggered once after a set delay.

Scheduler

The sceduler is the orchestrator that manages and runs the tasks. There is a NewScheduler function that creates a new scheduler which takes maxRunningTasks argument to limit the maximum number of tasks to run at a time. If the scheduler task queue is full, remaining tasks that are triggered will wait until there is a slot available in the scheduler. It has the following functionalities.

• AddTask: Registers a task to be run when triggered.
• RemoveTask: Removes a task from the scheduler. Tasks with only OneTimeTrigger will be removed automatically once run.
• Start: Starts the scheduler to monitor tasks.
• Stop: Stops the scheduler.

Documentation

Index

• type EventTrigger
  • func (t *EventTrigger) IsReady() bool
  • func (t *EventTrigger) Reset()
• type Execution
• type OneTimeTrigger
  • func (t *OneTimeTrigger) IsReady() bool
  • func (t *OneTimeTrigger) Reset()
• type PeriodicTrigger
  • func (t *PeriodicTrigger) IsReady() bool
  • func (t *PeriodicTrigger) Reset()
• type RetryPolicy
• type Scheduler
  • func NewScheduler(maxRunningTasks int) *Scheduler
  • func (s *Scheduler) AddTask(task *Task) *Task
  • func (s *Scheduler) RemoveTask(taskID int)
  • func (s *Scheduler) Start()
  • func (s *Scheduler) Stop()
• type Task
• type Trigger

Types

type EventTrigger

type EventTrigger struct {
	Trigger chan bool // Channel to signal an event.
}

EventTrigger triggers based on an external event signaled through a channel.

func (*EventTrigger) IsReady

func (t *EventTrigger) IsReady() bool

IsReady checks if there is a signal in the trigger channel.

func (*EventTrigger) Reset

func (t *EventTrigger) Reset()

Reset for EventTrigger does nothing as its state is managed externally.

type Execution

type Execution struct {
	StartedAt time.Time   // Start time of the execution.
	EndedAt   time.Time   // End time of the execution.
	Status    string      // Status of the execution (e.g., "SUCCESS", "FAILED").
	Error     string      // Error message if the execution failed.
	Event     interface{} // Event associated with the execution.
	Results   interface{} // Results of the execution.
}

Execution records the execution details of a task.

type OneTimeTrigger

type OneTimeTrigger struct {
	Delay time.Duration // The delay after which to trigger.
	// contains filtered or unexported fields
}

OneTimeTrigger triggers once after a specified delay.

func (*OneTimeTrigger) IsReady

func (t *OneTimeTrigger) IsReady() bool

IsReady checks if the current time has passed the delay period.

func (*OneTimeTrigger) Reset

func (t *OneTimeTrigger) Reset()

Reset sets the trigger registration time to the current time.

type PeriodicTrigger

type PeriodicTrigger struct {
	Interval time.Duration // Interval for periodic triggering.
	CronExpr string        // Cron expression for triggering.
	// contains filtered or unexported fields
}

PeriodicTrigger triggers at regular intervals or based on a cron expression.

func (*PeriodicTrigger) IsReady

func (t *PeriodicTrigger) IsReady() bool

IsReady checks if the trigger should activate based on time or cron expression.

func (*PeriodicTrigger) Reset

func (t *PeriodicTrigger) Reset()

Reset updates the last triggered time to the current time.

type RetryPolicy

type RetryPolicy struct {
	MaxRetries int           // Maximum number of retries.
	Delay      time.Duration // Delay between retries.
}

RetryPolicy defines the policy for retrying tasks on failure.

type Scheduler

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

Scheduler orchestrates the execution of tasks based on their triggers and priority.

func NewScheduler

func NewScheduler(maxRunningTasks int) *Scheduler

NewScheduler creates a new Scheduler with a specified limit on running tasks.

func (*Scheduler) AddTask

func (s *Scheduler) AddTask(task *Task) *Task

AddTask adds a new task to the scheduler and initializes its state.

func (*Scheduler) RemoveTask

func (s *Scheduler) RemoveTask(taskID int)

RemoveTask removes a task from the scheduler.

func (*Scheduler) Start

func (s *Scheduler) Start()

Start begins the scheduler's task execution loop.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop signals the scheduler to stop running tasks.

type Task

type Task struct {
	ID            int                          // Unique identifier for the task.
	Name          string                       // Name of the task.
	Description   string                       // Description of the task.
	Triggers      []Trigger                    // List of triggers for the task.
	Function      func(args interface{}) error // Function to execute as the task.
	Args          []interface{}                // Arguments for the task function.
	RetryPolicy   RetryPolicy                  // Retry policy for the task.
	Enabled       bool                         // Flag indicating if the task is enabled.
	Priority      int                          // Priority of the task for scheduling.
	ExecutionHist []Execution                  // History of task executions.
}

Task represents a schedulable task.

type Trigger

type Trigger interface {
	IsReady() bool // Returns true if the trigger condition is met.
	Reset()        // Resets the trigger state.
}

Trigger interface defines a method to check if a trigger condition is met.