machinery

Documentation

Index

• Constants
• func GetConn(p pool.Pool, timeout int) (*grpc.ClientConn, error)
• func GetPool(target string, min, max int) (pool.Pool, error)
• func GetQueueName(workerID, taskType string, reserved int) string
• func GetQueueNameByCluster(taskType, cluster string) string
• func InitMQInitMachineryServerWorker(queueName string, invoke Invoke, redis *redis.Config, concurrencyNum int, ...) (*machinery.Worker, error)
• func InitMachineryServer(queueName string, redis *redis.Config) (*machinery.Server, error)
• func ReturnConn(p pool.Pool, conn *grpc.ClientConn, err error) error
• func SetLoggerLevel(logger logging.ILogger)
• type Consumer
  • func NewConsumer(redis *redis.Config, errorsChan chan<- error, tracer opentracing.Tracer, ...) *Consumer
  • func (c *Consumer) Register(logicWorker IWorker, concurrentNum int) error
  • func (c *Consumer) Release()
  • func (c *Consumer) Start() error
  • func (c *Consumer) Stop()
• type ConsumerUnit
  • func NewConsumerUnit(logicWorker IWorker, redis *redis.Config, concurrentNum int, ...) (*ConsumerUnit, error)
  • func (unit *ConsumerUnit) Release() error
  • func (unit *ConsumerUnit) Resume(errorsChan chan<- error)
  • func (unit *ConsumerUnit) Start() error
  • func (unit *ConsumerUnit) Stop()
• type IWorker
• type Invoke
• type Logger
  • func NewMachineryLogger(logger logging.ILogger) *Logger
  • func (m *Logger) Fatal(args ...interface{})
  • func (m *Logger) Fatalf(str string, args ...interface{})
  • func (m *Logger) Fatalln(args ...interface{})
  • func (m *Logger) Panic(args ...interface{})
  • func (m *Logger) Panicf(str string, args ...interface{})
  • func (m *Logger) Panicln(args ...interface{})
  • func (m *Logger) Print(args ...interface{})
  • func (m *Logger) Printf(str string, args ...interface{})
  • func (m *Logger) Println(args ...interface{})

Constants

const (
	// PoolCount pool count
	PoolCount int = 2
	// PoolTypeName pool type name
	PoolTypeName = 0
	// PoolTypeHealth pool type health
	PoolTypeHealth = 1
)

Functions

func GetConn

func GetConn(p pool.Pool, timeout int) (*grpc.ClientConn, error)

GetConn get connection

func GetPool

func GetPool(target string, min, max int) (pool.Pool, error)

GetPool get pool

func GetQueueName

func GetQueueName(workerID, taskType string, reserved int) string

GetQueueName Get Queue Name

func GetQueueNameByCluster

func GetQueueNameByCluster(taskType, cluster string) string

GetQueueNameByCluster ...

func InitMQInitMachineryServerWorker

func InitMQInitMachineryServerWorker(
	queueName string, invoke Invoke, redis *redis.Config,
	concurrencyNum int, errorsChan chan<- error,
	tracer opentracing.Tracer, logger logging.ILogger,
) (*machinery.Worker, error)

InitMQInitMachineryServerWorker initialize the machinary message queue service work.

func InitMachineryServer

func InitMachineryServer(queueName string, redis *redis.Config) (*machinery.Server, error)

InitMachineryServer initialize Machinery service

func ReturnConn

func ReturnConn(p pool.Pool, conn *grpc.ClientConn, err error) error

ReturnConn return connection

func SetLoggerLevel

func SetLoggerLevel(logger logging.ILogger)

SetLoggerLevel set logger level

e.g.

if level.Level() > zapCore.DebugLevel {
	machinery.SetLoggerLevel(false)
}

Types

type Consumer

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

Consumer consumer

func NewConsumer

func NewConsumer(redis *redis.Config, errorsChan chan<- error, tracer opentracing.Tracer, logger logging.ILogger) *Consumer

NewConsumer create new consumer object

func (*Consumer) Register

func (c *Consumer) Register(logicWorker IWorker, concurrentNum int) error

Register register

func (*Consumer) Release

func (c *Consumer) Release()

Release release

func (*Consumer) Start

func (c *Consumer) Start() error

Start start

func (*Consumer) Stop

func (c *Consumer) Stop()

Stop stop

type ConsumerUnit

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

ConsumerUnit consumer unit

func NewConsumerUnit

func NewConsumerUnit(
	logicWorker IWorker, redis *redis.Config, concurrentNum int,
	errorsChan chan<- error, tracer opentracing.Tracer, logger logging.ILogger,
) (*ConsumerUnit, error)

NewConsumerUnit create a new consumer unit

func (*ConsumerUnit) Release

func (unit *ConsumerUnit) Release() error

Release release

func (*ConsumerUnit) Resume

func (unit *ConsumerUnit) Resume(errorsChan chan<- error)

Resume resume

func (*ConsumerUnit) Start

func (unit *ConsumerUnit) Start() error

Start startup

func (*ConsumerUnit) Stop

func (unit *ConsumerUnit) Stop()

Stop stop

type IWorker

type IWorker interface {
	Work(ctx context.Context, taskType, taskData string) (string, error)
	GetQueueName() string
	Start()
	Stop()
	Release() error
}

IWorker work interface

type Invoke

type Invoke func(ctx context.Context, taskType, taskData string) (string, error)

Invoke invoke

type Logger

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

Logger Logger

func NewMachineryLogger

func NewMachineryLogger(logger logging.ILogger) *Logger

NewMachineryLogger -

func (*Logger) Fatal

func (m *Logger) Fatal(args ...interface{})

Fatal -

func (*Logger) Fatalf

func (m *Logger) Fatalf(str string, args ...interface{})

Fatalf -

func (*Logger) Fatalln

func (m *Logger) Fatalln(args ...interface{})

Fatalln -

func (*Logger) Panic

func (m *Logger) Panic(args ...interface{})

Panic -

func (*Logger) Panicf

func (m *Logger) Panicf(str string, args ...interface{})

Panicf -

func (*Logger) Panicln

func (m *Logger) Panicln(args ...interface{})

Panicln -

func (*Logger) Print

func (m *Logger) Print(args ...interface{})

Print -

func (*Logger) Printf

func (m *Logger) Printf(str string, args ...interface{})

Printf -

func (*Logger) Println

func (m *Logger) Println(args ...interface{})

Println -