queue

Documentation

Index

• Constants
• func GetDuration(key interface{}) (time.Duration, error)
• type Bootstrapper
  • func (b *Bootstrapper) Bootstrap(context map[string]interface{}) error
• type Config
• type Server
  • func (qs *Server) EnqueueJob(taskName string, params map[string]interface{}) (TaskResult, error)
  • func (qs *Server) EnqueueJobWithMaxTries(taskName string, params map[string]interface{}) (TaskResult, error)
  • func (qs *Server) Name() string
  • func (qs *Server) RegisterTaskType(name string, task interface{})
  • func (qs *Server) Start(ctx context.Context, wg *sync.WaitGroup, startupErr chan<- error)
• type TaskQueuer
• type TaskResult
• type TaskType

Constants

const (
	TimeoutParam string = "Timeout"
)

Constants are commonly used by all the tasks through kwargs.

Functions

func GetDuration

func GetDuration(key interface{}) (time.Duration, error)

GetDuration parses key parameter to time.Duration type

Types

type Bootstrapper

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

Bootstrapper implements bootstrap.Bootstrapper.

func (*Bootstrapper) Bootstrap

func (b *Bootstrapper) Bootstrap(context map[string]interface{}) error

Bootstrap initiates the queue.

type Config

type Config interface {

	// GetNumWorkers gets the number of background workers to initiate
	GetNumWorkers() int

	// GetTaskRetries returns the number of retries allowed for a queued task
	GetTaskRetries() int

	// GetWorkerWaitTime gets the worker wait time for a task to be available while polling
	// increasing this may slow down task execution while reducing it may consume a lot of CPU cycles
	GetWorkerWaitTimeMS() int
}

Config is an interface for queue specific configurations

type Server

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

Server represents the queue server currently implemented based on gocelery

func (*Server) EnqueueJob

func (qs *Server) EnqueueJob(taskName string, params map[string]interface{}) (TaskResult, error)

EnqueueJob enqueues a job on the queue server for the given taskTypeName

func (*Server) EnqueueJobWithMaxTries

func (qs *Server) EnqueueJobWithMaxTries(taskName string, params map[string]interface{}) (TaskResult, error)

EnqueueJobWithMaxTries enqueues a job on the queue server for the given taskTypeName with maximum tries

func (*Server) Name

func (qs *Server) Name() string

Name of the queue server

func (*Server) RegisterTaskType

func (qs *Server) RegisterTaskType(name string, task interface{})

RegisterTaskType registers a task type on the queue server

func (*Server) Start

func (qs *Server) Start(ctx context.Context, wg *sync.WaitGroup, startupErr chan<- error)

Start the queue server

type TaskQueuer

type TaskQueuer interface {
	EnqueueJob(taskTypeName string, params map[string]interface{}) (TaskResult, error)
	EnqueueJobWithMaxTries(taskName string, params map[string]interface{}) (TaskResult, error)
}

TaskQueuer can be implemented by any queueing system

type TaskResult

type TaskResult interface {

	// Get the result within a timeout from the queue task execution
	Get(timeout time.Duration) (interface{}, error)
}

TaskResult represents a result from a queued task execution

type TaskType

type TaskType interface {

	// TaskTypeName of the task
	TaskTypeName() string
}

TaskType is a task to be queued in the centrifuge node to be completed asynchronously