tasks

Documentation

Overview

Package tasks provides workers that effectively run the instances of the scheduled jobs.

Index

• Constants
• Variables
• type Dispatcher
  • func NewDispatcher(maxWorkers int) *Dispatcher
  • func (d *Dispatcher) Run()
  • func (d *Dispatcher) Stop()
• type Runnable
  • func NewRunnable(ctx context.Context, cl client.Client, task *Task, action *jobs.Action, ...) Runnable
  • func RootRunnable(ctx context.Context, cl client.Client, task *Task) Runnable
  • func (r *Runnable) CreateChild(action *jobs.Action, message jobs.ActionMessage) Runnable
  • func (r *Runnable) Dispatch(input jobs.ActionMessage, actions []*jobs.Action, Queue chan Runnable)
  • func (r *Runnable) RunAction(Queue chan Runnable) error
• type Subscriber
  • func NewSubscriber(parentContext context.Context, client client.Client, server server.Server) *Subscriber
  • func (s *Subscriber) GetDispatcherForJob(job *jobs.Job) *Dispatcher
  • func (s *Subscriber) Init() error
  • func (s *Subscriber) ListenToMainQueue()
  • func (s *Subscriber) TaskChannelSubscription()
• type Task
  • func NewTaskFromEvent(ctx context.Context, job *jobs.Job, event interface{}) *Task
  • func (t *Task) Add(delta int)
  • func (t *Task) AppendLog(a jobs.Action, in jobs.ActionMessage, out jobs.ActionMessage)
  • func (t *Task) Done(delta int)
  • func (t *Task) EnqueueRunnables(c client.Client, output chan Runnable)
  • func (t *Task) GetJobTaskClone() *jobs.Task
  • func (t *Task) GetRunnableChannels() (*actions.RunnableChannels, chan bool)
  • func (t *Task) GlobalError(e error)
  • func (t *Task) Save()
  • func (t *Task) SetControllable(canStop bool, canPause bool)
  • func (t *Task) SetEndTime(ti time.Time)
  • func (t *Task) SetHasProgress()
  • func (t *Task) SetProgress(progress float32)
  • func (t *Task) SetStartTime(ti time.Time)
  • func (t *Task) SetStatus(status jobs.TaskStatus, message ...string)
• type Worker
  • func NewWorker(workerPool chan chan Runnable, requeue chan Runnable) Worker
  • func (w Worker) Start()
  • func (w Worker) Stop()

Constants

const (
	PubSubTopicTaskStatuses = "tasks"
	PubSubTopicControl      = "control"
)

const (
	// DefaultMaximumWorkers is set to 20.
	DefaultMaximumWorkers = 20
)

Variables

var (
	PubSub *pubsub.PubSub
)

var (
	RC int
)

Types

type Dispatcher

type Dispatcher struct {
	// A pool of workers channels that are registered with the dispatcher
	JobQueue   chan Runnable
	WorkerPool chan chan Runnable
	// contains filtered or unexported fields
}

Dispatcher orchestrates the jobs by dispatching work to available workers.

func NewDispatcher

func NewDispatcher(maxWorkers int) *Dispatcher

NewDispatcher creates and initialises a new Dispatcher with this amount of workers.

func (*Dispatcher) Run

func (d *Dispatcher) Run()

Run simply starts the N workers of this dispacher.

func (*Dispatcher) Stop

func (d *Dispatcher) Stop()

Stop sends a quit signal to all workers. NOT YET IMPLEMENTED

type Runnable

type Runnable struct {
	jobs.Action
	Task           *Task
	Message        jobs.ActionMessage
	Client         client.Client
	Context        context.Context
	Implementation actions.ConcreteAction
}

Runnable represents the runnable instance of a given task

func NewRunnable

func NewRunnable(ctx context.Context, cl client.Client, task *Task, action *jobs.Action, message jobs.ActionMessage) Runnable

NewRunnable creates a new runnable and populates it with the concrete task implementation found with action.ID, if such an implementation is found.

func RootRunnable

func RootRunnable(ctx context.Context, cl client.Client, task *Task) Runnable

func (*Runnable) CreateChild

func (r *Runnable) CreateChild(action *jobs.Action, message jobs.ActionMessage) Runnable

CreateChild replicates a runnable for child action

func (*Runnable) Dispatch

func (r *Runnable) Dispatch(input jobs.ActionMessage, actions []*jobs.Action, Queue chan Runnable)

Dispatch gets next runnable from Action and enqueues it to the Queue Todo - Check that done channel is working correctly with chained actions

func (*Runnable) RunAction

func (r *Runnable) RunAction(Queue chan Runnable) error

RunAction creates an action and calls Dispatch

type Subscriber

type Subscriber struct {
	Client          client.Client
	MainQueue       chan Runnable
	UpdateTasksChan chan *jobs.Task

	JobsDefinitions map[string]*jobs.Job
	Dispatchers     map[string]*Dispatcher

	RootContext context.Context
	// contains filtered or unexported fields
}

Subscriber handles incoming events, applies selectors if any and generates all ActionMessage to trigger actions

func NewSubscriber

func NewSubscriber(parentContext context.Context, client client.Client, server server.Server) *Subscriber

NewSubscriber creates a multiplexer for tasks managements and messages by maintaining a map of dispacher, one for each job definition.

func (*Subscriber) GetDispatcherForJob

func (s *Subscriber) GetDispatcherForJob(job *jobs.Job) *Dispatcher

GetDispatcherForJob creates a new dispatcher for a job

func (*Subscriber) Init

func (s *Subscriber) Init() error

Init subscriber with current list of jobs from Jobs service

func (*Subscriber) ListenToMainQueue

func (s *Subscriber) ListenToMainQueue()

ListenToMainQueue starts a go routine that listens to the Event Bus

func (*Subscriber) TaskChannelSubscription

func (s *Subscriber) TaskChannelSubscription()

TaskChannelSubscription uses PubSub library to receive update messages from tasks

type Task

type Task struct {
	*jobs.Job

	RC int
	// contains filtered or unexported fields
}

func NewTaskFromEvent

func NewTaskFromEvent(ctx context.Context, job *jobs.Job, event interface{}) *Task

func (*Task) Add

func (t *Task) Add(delta int)

func (*Task) AppendLog

func (t *Task) AppendLog(a jobs.Action, in jobs.ActionMessage, out jobs.ActionMessage)

func (*Task) Done

func (t *Task) Done(delta int)

func (*Task) EnqueueRunnables

func (t *Task) EnqueueRunnables(c client.Client, output chan Runnable)

func (*Task) GetJobTaskClone

func (t *Task) GetJobTaskClone() *jobs.Task

func (*Task) GetRunnableChannels

func (t *Task) GetRunnableChannels() (*actions.RunnableChannels, chan bool)

func (*Task) GlobalError

func (t *Task) GlobalError(e error)

func (*Task) Save

func (t *Task) Save()

func (*Task) SetControllable

func (t *Task) SetControllable(canStop bool, canPause bool)

func (*Task) SetEndTime

func (t *Task) SetEndTime(ti time.Time)

func (*Task) SetHasProgress

func (t *Task) SetHasProgress()

func (*Task) SetProgress

func (t *Task) SetProgress(progress float32)

func (*Task) SetStartTime

func (t *Task) SetStartTime(ti time.Time)

func (*Task) SetStatus

func (t *Task) SetStatus(status jobs.TaskStatus, message ...string)

type Worker

type Worker struct {
	WorkerPool chan chan Runnable
	JobChannel chan Runnable

	JobReQueue chan Runnable
	// contains filtered or unexported fields
}

Worker represents the worker that executes the jobs.

func NewWorker

func NewWorker(workerPool chan chan Runnable, requeue chan Runnable) Worker

NewWorker creates and configures a new worker.

func (Worker) Start

func (w Worker) Start()

Start method starts the run loop for the worker, listening for a quit channel in case we need to stop it.

func (Worker) Stop

func (w Worker) Stop()

Stop signals the worker to stop listening for work requests.