coda

Documentation

Index

• Constants
• Variables
• type Agent
  • func NewAgent(t TaskManager, 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 Task
  • func NewTask() Task
  • func (s Task) GetMeta(name string) interface{}
  • func (s Task) Save(ts TaskSaver)
  • func (s Task) SetMeta(name string, value interface{}) Task
• type TaskManager
• type TaskManagerMongo
  • func NewTaskManagerMongo(taskCollection collection) (tm *TaskManagerMongo)
  • func (s *TaskManagerMongo) FindAndStallTaskForCaller(callerName string) (task Task, err error)
  • func (s *TaskManagerMongo) FindTask(id string) (t Task, err error)
  • func (s *TaskManagerMongo) GarbageCollectExpiredAgents(callerName string) (changeInfo *mgo.ChangeInfo, err error)
  • func (s *TaskManagerMongo) NewTask(callerName string, profile ProfileType, status string) (t Task)
  • func (s *TaskManagerMongo) SaveTask(t Task) (Task, error)
  • func (s *TaskManagerMongo) ScheduleTask(task Task, expireTime time.Time) Task
  • func (s *TaskManagerMongo) SubscribeToSchedule(callerName string) (subscription chan Task)
• type TaskSaver

Constants

const (
	//TaskAgentLongRunning --
	TaskAgentLongRunning ProfileType = "agent_task_long_running"
	//TaskAgentScheduledTask --
	TaskAgentScheduledTask ProfileType = "agent_scheduled_task"
	//TaskLongPollQueue --
	TaskLongPollQueue ProfileType = "longpoll_queue"

	//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(2) * 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 TaskManager, 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 Task

type Task struct {
	ID         bson.ObjectId          `bson:"_id"`
	Timestamp  time.Time              `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"`
}

Task - a task object

func NewTask

func NewTask() Task

NewTask - a helper to create a new task

func (Task) GetMeta

func (s Task) GetMeta(name string) interface{}

GetMeta - get a public metadata record

func (Task) Save

func (s Task) Save(ts TaskSaver)

Save -- Safe way to update a task

func (Task) SetMeta

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

SetMeta - set a public metadata record

type TaskManager

type TaskManager 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) Task
}

TaskManager ---

type TaskManagerMongo

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

TaskManagerMongo - manages task interactions crud stuff

func NewTaskManagerMongo

func NewTaskManagerMongo(taskCollection collection) (tm *TaskManagerMongo)

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

func (*TaskManagerMongo) FindAndStallTaskForCaller

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

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

func (*TaskManagerMongo) FindTask

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

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

func (*TaskManagerMongo) GarbageCollectExpiredAgents

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

GarbageCollectExpiredAgents --

func (*TaskManagerMongo) NewTask

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

NewTask - get us a new empty task

func (*TaskManagerMongo) SaveTask

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

SaveTask - saves the given task

func (*TaskManagerMongo) ScheduleTask

func (s *TaskManagerMongo) ScheduleTask(task Task, expireTime time.Time) Task

ScheduleTask --

func (*TaskManagerMongo) SubscribeToSchedule

func (s *TaskManagerMongo) 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 TaskSaver

type TaskSaver interface {
	SaveTask(t Task) (Task, error)
}

TaskSaver - interface that can save a task