machinery

Documentation

Index

• type Server
  • func NewServer(tasks map[string]interface{}, configPaths ...interface{}) (*Server, error)
  • func (s *Server) ParseConfig(configPaths ...interface{}) (err error)
  • func (s *Server) Send(ctx context.Context, taskSignatures ...*tasks.Signature) ([]*result.AsyncResult, error)
  • func (s *Server) SendChain(ctx context.Context, taskSignatures ...*tasks.Signature) (*result.ChainAsyncResult, error)
  • func (s *Server) SendGroup(ctx context.Context, sendConcurrency int, taskSignatures ...*tasks.Signature) ([]*result.AsyncResult, error)
  • func (s *Server) SendGroupWithChord(ctx context.Context, sendConcurrency int, chordTask *tasks.Signature, ...) (*result.ChordAsyncResult, error)
  • func (s *Server) StartWorker(options ...WorkerOption) error
• type Task
  • func NewTask(name string, args ...TaskArg) *Task
• type TaskArg
  • func Arg(dataType string, dataValue interface{}, name ...string) TaskArg
• type TaskArgs
  • func Args() TaskArgs
  • func (t *TaskArgs) Add(dataType string, dataValue interface{}, name ...string) *TaskArgs
  • func (t *TaskArgs) AddArg(args ...TaskArg) *TaskArgs
  • func (t *TaskArgs) Slice() TaskArgs
• type WorkerConfig
• type WorkerOption
  • func Concurrency(concurrency int) WorkerOption
  • func ConsumerTag(tag string) WorkerOption
  • func ErrorHandler(errorHandler func(err error)) WorkerOption
  • func PostTaskHandler(postTaskHandler func(signature *tasks.Signature)) WorkerOption
  • func PreTaskHandler(preTaskHandler func(signature *tasks.Signature)) WorkerOption

Types

type Server

type Server struct {
	*machinery.Server
	// contains filtered or unexported fields
}

func NewServer

func NewServer(tasks map[string]interface{}, configPaths ...interface{}) (*Server, error)

func (*Server) ParseConfig

func (s *Server) ParseConfig(configPaths ...interface{}) (err error)

func (*Server) Send

func (s *Server) Send(ctx context.Context, taskSignatures ...*tasks.Signature) ([]*result.AsyncResult, error)

Send 发送到队列

func (*Server) SendChain

func (s *Server) SendChain(ctx context.Context, taskSignatures ...*tasks.Signature) (*result.ChainAsyncResult, error)

func (*Server) SendGroup

func (s *Server) SendGroup(ctx context.Context, sendConcurrency int, taskSignatures ...*tasks.Signature) ([]*result.AsyncResult, error)

func (*Server) SendGroupWithChord

func (s *Server) SendGroupWithChord(ctx context.Context, sendConcurrency int, chordTask *tasks.Signature, taskSignatures ...*tasks.Signature) (*result.ChordAsyncResult, error)

func (*Server) StartWorker

func (s *Server) StartWorker(options ...WorkerOption) error

type Task

type Task = tasks.Signature

func NewTask

func NewTask(name string, args ...TaskArg) *Task

NewTask creates a new task signature

type TaskArg

type TaskArg = tasks.Arg

func Arg

func Arg(dataType string, dataValue interface{}, name ...string) TaskArg

type TaskArgs

type TaskArgs []TaskArg

func Args

func Args() TaskArgs

func (*TaskArgs) Add

func (t *TaskArgs) Add(dataType string, dataValue interface{}, name ...string) *TaskArgs

func (*TaskArgs) AddArg

func (t *TaskArgs) AddArg(args ...TaskArg) *TaskArgs

func (*TaskArgs) Slice

func (t *TaskArgs) Slice() TaskArgs

type WorkerConfig

type WorkerConfig struct {
	ConsumerTag string
	Concurrency int
	// contains filtered or unexported fields
}

type WorkerOption

type WorkerOption func(*WorkerConfig)

func Concurrency

func Concurrency(concurrency int) WorkerOption

func ConsumerTag

func ConsumerTag(tag string) WorkerOption

func ErrorHandler

func ErrorHandler(errorHandler func(err error)) WorkerOption

func PostTaskHandler

func PostTaskHandler(postTaskHandler func(signature *tasks.Signature)) WorkerOption

func PreTaskHandler

func PreTaskHandler(preTaskHandler func(signature *tasks.Signature)) WorkerOption