mq

Documentation

Index

• func Prefix(key string, prefix string) string
• type Backend
• type Broker
• type Canceller
• type ChannelMgr
• type CronDescriber
• type Job
• type JobQueue
  • func NewJobQueue(broker Broker, backend Backend) *JobQueue
  • func (mq *JobQueue) Cancel(id string) error
  • func (mq *JobQueue) ListChannel() ([]string, error)
  • func (mq *JobQueue) ListSubject(channel string) ([]string, error)
  • func (mq *JobQueue) Next(channel string, subject string, data []byte) (*Task, error)
  • func (mq *JobQueue) Publish(channel string, subject string, data []byte) (*Task, error)
  • func (mq *JobQueue) Register(subject string, job Job)
  • func (mq *JobQueue) RegisterReceiver(receiver func(*TaskStatus) error)
  • func (mq *JobQueue) SendTask(task *Task) error
  • func (mq *JobQueue) StartReceiver(numWorkers int)
  • func (mq *JobQueue) StartWorker(channel string, numWorkers int)
  • func (mq *JobQueue) StopReceiver()
  • func (mq *JobQueue) StopWorker()
• type MsgHead
• type Status
• type Task
  • func NewTask(channel string, subject string, data []byte) *Task
  • func (task *Task) Cancelled() *TaskStatus
  • func (task *Task) Failed(traceback error) *TaskStatus
  • func (task *Task) Pending() *TaskStatus
  • func (task *Task) Running() *TaskStatus
  • func (task *Task) Success(result interface{}) *TaskStatus
• type TaskStatus
• type Worker
  • func NewWorker(process func() error, numWorkers int) *Worker
  • func (mq *Worker) Start()
  • func (mq *Worker) Stop()

Functions

func Prefix

func Prefix(key string, prefix string) string

Types

type Backend

type Backend interface {
	Canceller
	ChannelMgr
	FeedBack(taskStatus *TaskStatus) (err error)
	GetFeedback() (taskStatus *TaskStatus, err error)
}

type Broker

type Broker interface {
	SendTask(task *Task) (err error)
	GetTask(channel string) (task *Task, err error)
}

type Canceller

type Canceller interface {
	Cancel(id string) (err error)
	IsCancelled(id string) (ok bool, err error)
	ClearCancellation(id string) (err error)
}

type ChannelMgr

type ChannelMgr interface {
	ListChannel() (channelList []string, err error)
	ListSubject(channel string) (subjectList []string, err error)
	RegisterChannel(channel string, subjectList []string) error
}

type CronDescriber

type CronDescriber interface {
	CronSpec() string
}

type Job

type Job = func(task *Task) (result interface{}, err error)

type JobQueue

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

func NewJobQueue

func NewJobQueue(broker Broker, backend Backend) *JobQueue

func (*JobQueue) Cancel

func (mq *JobQueue) Cancel(id string) error

func (*JobQueue) ListChannel

func (mq *JobQueue) ListChannel() ([]string, error)

func (*JobQueue) ListSubject

func (mq *JobQueue) ListSubject(channel string) ([]string, error)

func (*JobQueue) Next

func (mq *JobQueue) Next(channel string, subject string, data []byte) (*Task, error)

func (*JobQueue) Publish

func (mq *JobQueue) Publish(channel string, subject string, data []byte) (*Task, error)

func (*JobQueue) Register

func (mq *JobQueue) Register(subject string, job Job)

func (*JobQueue) RegisterReceiver

func (mq *JobQueue) RegisterReceiver(receiver func(*TaskStatus) error)

func (*JobQueue) SendTask

func (mq *JobQueue) SendTask(task *Task) error

func (*JobQueue) StartReceiver

func (mq *JobQueue) StartReceiver(numWorkers int)

func (*JobQueue) StartWorker

func (mq *JobQueue) StartWorker(channel string, numWorkers int)

func (*JobQueue) StopReceiver

func (mq *JobQueue) StopReceiver()

func (*JobQueue) StopWorker

func (mq *JobQueue) StopWorker()

type MsgHead

type MsgHead struct {
	Channel string `json:"channel"`
	Subject string `json:"subject"`
	ID      string `json:"id"`
}

type Status

type Status string

const (
	StatusPending   Status = "PENDING"
	StatusRunning   Status = "RUNNING"
	StatusCancelled Status = "CANCELED"
	StatusSuccess   Status = "SUCCESS"
	StatusFailed    Status = "FAILED"
)

type Task

type Task struct {
	MsgHead
	Data      []byte    `json:"data,omitempty"`
	CreatedAt time.Time `json:"created_at"`
}

func NewTask

func NewTask(channel string, subject string, data []byte) *Task

func (*Task) Cancelled

func (task *Task) Cancelled() *TaskStatus

func (*Task) Failed

func (task *Task) Failed(traceback error) *TaskStatus

func (*Task) Pending

func (task *Task) Pending() *TaskStatus

func (*Task) Running

func (task *Task) Running() *TaskStatus

func (*Task) Success

func (task *Task) Success(result interface{}) *TaskStatus

type TaskStatus

type TaskStatus struct {
	MsgHead
	Status    Status    `json:"status"`
	UpdatedAt time.Time `json:"updated_at"`
	Args      []byte    `json:"args,omitempty"`
	Result    []byte    `json:"result,omitempty"`
	Traceback []byte    `json:"traceback,omitempty"`
}

type Worker

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

func NewWorker

func NewWorker(process func() error, numWorkers int) *Worker

func (*Worker) Start

func (mq *Worker) Start()

func (*Worker) Stop

func (mq *Worker) Stop()