taskmanager

Documentation

Index

• Constants
• Variables
• type Agent
  • func NewAgent(t TaskManagerInterface, callerName string) *Agent
  • func (s *Agent) GetStatus() chan string
  • func (s *Agent) GetTask() *Task
  • func (s *Agent) Run(process func(*Agent) error)
• type ProfileType
• type RedactedTask
• type Task
  • func (s *Task) GetPrivateMeta(name string) interface{}
  • func (s *Task) GetPublicMeta(name string) interface{}
  • func (s *Task) GetRedactedVersion() RedactedTask
  • func (s *Task) SetPrivateMeta(name string, value interface{})
  • func (s *Task) SetPublicMeta(name string, value interface{})
• type TaskManager
  • func NewTaskManager(taskCollection integrations.Collection) (tm *TaskManager)
  • func (s *TaskManager) FindAndStallTaskForCaller(callerName string) (task *Task, err error)
  • func (s *TaskManager) FindTask(id string) (t *Task, err error)
  • func (s *TaskManager) GarbageCollectExpiredAgents(callerName string) (changeInfo *mgo.ChangeInfo, err error)
  • func (s *TaskManager) NewTask(callerName string, profile ProfileType, status string) (t *Task)
  • func (s *TaskManager) SaveTask(t *Task) (*Task, error)
  • func (s *TaskManager) ScheduleTask(t *Task, expireTime time.Time)
  • func (s *TaskManager) SubscribeToSchedule(callerName string) (subscription chan *Task)
• type TaskManagerInterface

Constants

const (
	//TaskAgentLongRunning --
	TaskAgentLongRunning ProfileType = "agent_task_long_running"
	//TaskAgentScheduledTask --
	TaskAgentScheduledTask ProfileType = "agent_scheduled_task"
	//TaskLeaseProcurement --
	TaskLeaseProcurement ProfileType = "lease_procurement"
	//TaskLeaseReStock --
	TaskLeaseReStock ProfileType = "lease_restock"
	//TaskInventoryLedger --
	TaskInventoryLedger ProfileType = "inventory_ledger"
	//TaskLongPollQueue --
	TaskLongPollQueue ProfileType = "longpoll_queue"
	//TaskChildID -- child task spawned from current task
	TaskChildID = "child_task_id"
	//TaskActionMetaName --
	TaskActionMetaName = "task_action"

	//ExpiredTask -
	ExpiredTask int64 = 0
	//TaskStatusAvailable --- task status is set to available
	TaskStatusAvailable = "available"

	//AgentTaskStatusInitializing ---
	AgentTaskStatusInitializing = "initializing"
	//AgentTaskStatusScheduled ---
	AgentTaskStatusScheduled = "scheduled"
	//AgentTaskStatusRunning ---
	AgentTaskStatusRunning = "running"
	//AgentTaskStatusComplete ---
	AgentTaskStatusComplete = "complete"
	//AgentTaskStatusFailed ---
	AgentTaskStatusFailed = "failed"
)

Variables

var (
	//ErrNoResults - no results found in query
	ErrNoResults = errors.New("no results found")
	//AgentTaskPollerInterval - time offset to poll a task from an agent
	AgentTaskPollerInterval = time.Duration(30) * time.Second
	//AgentTaskPollerTimeout - time until a agent will expire its task if not polled
	AgentTaskPollerTimeout = time.Duration(5) * time.Minute
)

Types

type Agent

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

Agent an object which knows how to handle long running tasks. polling, timeouts etc

func NewAgent

func NewAgent(t TaskManagerInterface, callerName string) *Agent

NewAgent -- creates a new initialized agent object

func (*Agent) GetStatus

func (s *Agent) GetStatus() chan string

GetStatus - returns a status emitting channel

func (*Agent) GetTask

func (s *Agent) GetTask() *Task

GetTask - get the agents task object

func (*Agent) Run

func (s *Agent) Run(process func(*Agent) error)

Run - this begins the running of an agent's async process

type ProfileType

type ProfileType string

ProfileType - indicator of the purpose of the task to be performed

type RedactedTask

type RedactedTask struct {
	ID         bson.ObjectId          `bson:"_id"`
	Timestamp  int64                  `bson:"timestamp"`
	Expires    int64                  `bson:"expires"`
	Status     string                 `bson:"status"`
	Profile    ProfileType            `bson:"profile"`
	CallerName string                 `bson:"caller_name"`
	MetaData   map[string]interface{} `bson:"metadata"`
}

RedactedTask - a task object without sensitive information

type Task

type Task struct {
	ID              bson.ObjectId          `bson:"_id"`
	Timestamp       int64                  `bson:"timestamp"`
	Expires         int64                  `bson:"expires"`
	Status          string                 `bson:"status"`
	Profile         ProfileType            `bson:"profile"`
	CallerName      string                 `bson:"caller_name"`
	MetaData        map[string]interface{} `bson:"metadata"`
	PrivateMetaData map[string]interface{} `bson:"private_metadata"`
}

Task - a task object

func (*Task) GetPrivateMeta

func (s *Task) GetPrivateMeta(name string) interface{}

GetPrivateMeta - get a private meta record

func (*Task) GetPublicMeta

func (s *Task) GetPublicMeta(name string) interface{}

GetPublicMeta - get a public metadata record

func (*Task) GetRedactedVersion

func (s *Task) GetRedactedVersion() RedactedTask

GetRedactedVersion - returns a redacted version of this task, removing private info

func (*Task) SetPrivateMeta

func (s *Task) SetPrivateMeta(name string, value interface{})

SetPrivateMeta - set a private meta data record

func (*Task) SetPublicMeta

func (s *Task) SetPublicMeta(name string, value interface{})

SetPublicMeta - set a public metadata record

type TaskManager

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

TaskManager - manages task interactions crud stuff

func NewTaskManager

func NewTaskManager(taskCollection integrations.Collection) (tm *TaskManager)

NewTaskManager - this creates a new task manager object and returns it

func (*TaskManager) FindAndStallTaskForCaller

func (s *TaskManager) FindAndStallTaskForCaller(callerName string) (task *Task, err error)

FindAndStallTaskForCaller - find and lock the first matching task, then return it

func (*TaskManager) FindTask

func (s *TaskManager) FindTask(id string) (t *Task, err error)

FindTask - this will find and return a task with a given ID

func (*TaskManager) GarbageCollectExpiredAgents

func (s *TaskManager) GarbageCollectExpiredAgents(callerName string) (changeInfo *mgo.ChangeInfo, err error)

func (*TaskManager) NewTask

func (s *TaskManager) NewTask(callerName string, profile ProfileType, status string) (t *Task)

NewTask - get us a new empty task

func (*TaskManager) SaveTask

func (s *TaskManager) SaveTask(t *Task) (*Task, error)

SaveTask - saves the given task

func (*TaskManager) ScheduleTask

func (s *TaskManager) ScheduleTask(t *Task, expireTime time.Time)

func (*TaskManager) SubscribeToSchedule

func (s *TaskManager) SubscribeToSchedule(callerName string) (subscription chan *Task)

SubscribeToSchedule - subscribe to a schedule and get a channel to listen on for a task when it hits its scheduled time

type TaskManagerInterface

type TaskManagerInterface interface {
	NewTask(callerName string, profile ProfileType, status string) (t *Task)
	FindTask(id string) (t *Task, err error)
	FindAndStallTaskForCaller(callerName string) (task *Task, err error)
	SaveTask(t *Task) (*Task, error)
	ScheduleTask(t *Task, expireTime time.Time)
}

TaskManagerInterface ---