queue

Documentation

Index

• Constants
• Variables
• func DeregisterProducer(name string)
• func DeregisterrConsumer(name string)
• func NewBuilder() container.StandardBuilder
• func RegisterConsumer(resolver mqcResover)
• func RegisterProducer(resolver imqpResover)
• type ConsumeCallback
• type IComponentQueue
• type IMQC
  • func NewMQC(proto string, setting config.Config) (IMQC, error)
• type IMQCMessage
• type IMQP
  • func NewMQP(proto string, setting config.Config) (IMQP, error)
• type IQueue
• type Message
  • func NewMsg(obj interface{}, opts ...Option) (msg Message, err error)
• type MsgItem
  • func (w *MsgItem) Body() map[string]interface{}
  • func (w *MsgItem) Header() map[string]string
  • func (w *MsgItem) String() string
• type MsgWrap
  • func (w *MsgWrap) Body() map[string]interface{}
  • func (w *MsgWrap) Header() map[string]string
  • func (w *MsgWrap) String() string
• type Option
  • func WithHeader(key, val string) Option
  • func WithXRequestID(reqId string) Option
• type StandardQueue
  • func NewStandardQueue(c container.Container) StandardQueue
• type TaskInfo

Constants

const (
	TypeNode = "queues"
)

Variables

var DefaultMaxQueueLen = 100

默认最大队列长度 500

var Nil error = errors.New("Queue Nil")

Functions

func DeregisterProducer

func DeregisterProducer(name string)

Deregister 清理配置适配器

func DeregisterrConsumer

func DeregisterrConsumer(name string)

Deregister 清理配置适配器

func NewBuilder

func NewBuilder() container.StandardBuilder

func RegisterConsumer

func RegisterConsumer(resolver mqcResover)

RegisterConsumer 注册消息消费

func RegisterProducer

func RegisterProducer(resolver imqpResover)

RegisterProducer 注册配置文件适配器

Types

type ConsumeCallback

type ConsumeCallback func(IMQCMessage)

type IComponentQueue

type IComponentQueue interface {
	GetQueue(name string) (q IQueue)
}

IComponentQueue Component Queue

type IMQC

type IMQC interface {
	Connect() error
	Consume(task TaskInfo, callback ConsumeCallback) (err error)
	Unconsume(queue string)
	Start()
	Close()
}

IMQC consumer接口

func NewMQC

func NewMQC(proto string, setting config.Config) (IMQC, error)

NewMQC 根据适配器名称及参数返回配置处理器

type IMQCMessage

type IMQCMessage interface {
	RetryCount() int64
	Ack() error
	Nack(error) error
	Original() string
	GetMessage() Message
}

IMQCMessage 队列消息

type IMQP

type IMQP interface {
	Push(key string, value Message) error
	DelayPush(key string, value Message, delaySeconds int64) error
	Count(key string) (int64, error)
	Close() error
}

IMQP 消息生产

func NewMQP

func NewMQP(proto string, setting config.Config) (IMQP, error)

NewMQP 根据适配器名称及参数返回配置处理器

type IQueue

type IQueue interface {
	Send(ctx context.Context, key string, value interface{}) error
	DelaySend(ctx context.Context, key string, value interface{}, delaySeconds int64) error
	Count(key string) (int64, error)
}

IQueue 消息队列

type Message

type Message interface {
	Header() map[string]string
	Body() map[string]interface{}
	String() string
}

func NewMsg

func NewMsg(obj interface{}, opts ...Option) (msg Message, err error)

type MsgItem

type MsgItem struct {
	HeaderMap xtypes.SMap `json:"header"`
	BodyMap   xtypes.XMap `json:"body"`
	// contains filtered or unexported fields
}

func (*MsgItem) Body

func (w *MsgItem) Body() map[string]interface{}

func (*MsgItem) Header

func (w *MsgItem) Header() map[string]string

func (*MsgItem) String

func (w *MsgItem) String() string

type MsgWrap

type MsgWrap struct {
	HeaderMap xtypes.SMap `json:"header,omitempty"`
	BodyMap   xtypes.XMap `json:"body"`
	// contains filtered or unexported fields
}

func (*MsgWrap) Body

func (w *MsgWrap) Body() map[string]interface{}

func (*MsgWrap) Header

func (w *MsgWrap) Header() map[string]string

func (*MsgWrap) String

func (w *MsgWrap) String() string

type Option

type Option func(m *MsgWrap)

func WithHeader

func WithHeader(key, val string) Option

func WithXRequestID

func WithXRequestID(reqId string) Option

type StandardQueue

type StandardQueue interface {
	GetQueue(name ...string) (q IQueue)
}

func NewStandardQueue

func NewStandardQueue(c container.Container) StandardQueue

NewStandardQueue 创建queue

type TaskInfo

type TaskInfo interface {
	GetQueue() string
	GetConcurrency() int
}