taskqueue

Documentation

Index

• type AsyncJobMsg
• type AsyncJobQueueKey
• type AsyncJobState
• type QueueManager
  • func NewQueueManager() *QueueManager
  • func (q *QueueManager) GetTaskState(brokerid string, instanceid string, operation base.Operation) (*AsyncJobState, error)
  • func (q *QueueManager) Init()
  • func (q *QueueManager) IsTaskScheduled(id string) bool
  • func (q *QueueManager) RequestTaskQueue(brokerid string, instanceid string, operation base.Operation) (chan AsyncJobMsg, error)
  • func (q *QueueManager) ScheduleTask(cronExpression string, id string, task interface{}) (*gocron.Job, error)
  • func (q *QueueManager) UnScheduleTask(id string) error

Types

type AsyncJobMsg

type AsyncJobMsg struct {
	BrokerId   string
	InstanceId string
	JobType    base.Operation
	JobState   AsyncJobState
}

messages of job state delivered over chan that are persisited

type AsyncJobQueueKey

type AsyncJobQueueKey struct {
	BrokerId   string
	InstanceId string
	Operation  base.Operation
}

Jobs are unique for a broker,instance, and operation (CreateOp,DeleteOp,ModifyOp, BindOp, UnBindOp) this identifier is used as the unique key to retrieve a chan and or job state

type AsyncJobState

type AsyncJobState struct {
	State   base.InstanceState
	Message string
}

job state object persisted for brokers to access

type QueueManager

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

QueueManager maintains:

 	A set of open channels for active jobs
	A list of jobstates for requested job
	A task scheduler for cleanup of jobstates
	A list of jobstates that need cleanup

func NewQueueManager

func NewQueueManager() *QueueManager

can be called to initialize the manager defaults to do clean-up of jobstates after an hour. runs clean up check every 15 minutes

func (*QueueManager) GetTaskState

func (q *QueueManager) GetTaskState(brokerid string, instanceid string, operation base.Operation) (*AsyncJobState, error)

a broker or adapter can query the state of a job, will return an error if there is no known state. jobstates get cleaned-up automatically after a period of time after the chan is closed we cant do clean up here because state means different things to different brokers

func (*QueueManager) Init

func (q *QueueManager) Init()

must be called to activate cleanup mechanism separated from constructor to allow config and testing

func (*QueueManager) IsTaskScheduled

func (q *QueueManager) IsTaskScheduled(id string) bool

Determine if job(id) is scheduled

func (*QueueManager) RequestTaskQueue

func (q *QueueManager) RequestTaskQueue(brokerid string, instanceid string, operation base.Operation) (chan AsyncJobMsg, error)

a broker or adapter can request a channel to communicate state of async processes. the queue manager will launch a channel monitor to recieve messages and update the state of that async operation. will return an error if a channel has already been launched. Channels must be closed by the recipient after use. job state will be persisted and retained for a period of time before being cleaned up.

func (*QueueManager) ScheduleTask

func (q *QueueManager) ScheduleTask(cronExpression string, id string, task interface{}) (*gocron.Job, error)

Allow Jobs to be scheduled by brokers

func (*QueueManager) UnScheduleTask

func (q *QueueManager) UnScheduleTask(id string) error

Stop jobs scheduled