mq

Documentation

Index

• type AsyncSendCallback
• type CompressionCodec
• type ConsumerConfig
• type ConsumerRegistry
• type IConsumer
• type IConsumerHandler
• type IConsumerWorker
• type IFactory
• type IProducer
• type Message
  • func NewDeferMessage(topic string, deferDuration time.Duration, body []byte) *Message
  • func NewMessage(topic string, body []byte) *Message
  • func NewOrderlyMessage(topic, partitionKey string, body []byte) *Message
  • func (m *Message) WithDeferDuration(deferDuration time.Duration) *Message
  • func (m *Message) WithPartitionKey(partitionKey string) *Message
  • func (m *Message) WithProperties(properties map[string]string) *Message
  • func (m *Message) WithTag(tag string) *Message
• type MessageExt
• type ProducerConfig
• type SendResponse

Types

type AsyncSendCallback

type AsyncSendCallback func(ctx context.Context, sendResponse SendResponse, err error)

type CompressionCodec

type CompressionCodec int

const (
	// CompressionNone no compression
	CompressionNone CompressionCodec = iota
	// CompressionZSTD compression using ZSTD
	CompressionZSTD
)

type ConsumerConfig

type ConsumerConfig struct {
	// Name server address
	Addr []string
	// Topic name
	Topic string
	// Consumer group name
	ConsumerGroup string
	// Whether to consume orderly
	Orderly bool
	// Consume specific tags, such as "tag" or "tag1 || tag2 || tag3"
	TagExpression string
	// Max number of messages consumed concurrently
	ConsumeGoroutineNums int
	// Timeout for consumer one message
	ConsumeTimeout time.Duration
}

type ConsumerRegistry

type ConsumerRegistry interface {
	Register(worker []IConsumerWorker) ConsumerRegistry
	StartAll(ctx context.Context) error
}

type IConsumer

type IConsumer interface {
	Start() error
	Close() error
	RegisterHandler(IConsumerHandler)
}

type IConsumerHandler

type IConsumerHandler interface {
	HandleMessage(context.Context, *MessageExt) error
}

type IConsumerWorker

type IConsumerWorker interface {
	ConsumerCfg(ctx context.Context) (*ConsumerConfig, error)
	IConsumerHandler
}

type IFactory

type IFactory interface {
	NewProducer(ProducerConfig) (IProducer, error)
	NewConsumer(ConsumerConfig) (IConsumer, error)
}

type IProducer

type IProducer interface {
	Start() error
	Close() error
	Send(ctx context.Context, message *Message) (SendResponse, error)
	SendBatch(ctx context.Context, messages []*Message) (SendResponse, error)
	SendAsync(ctx context.Context, callback AsyncSendCallback, message *Message) error
}

type Message

type Message struct {
	Topic         string
	Body          []byte
	Tag           string
	PartitionKey  string
	Properties    map[string]string
	DeferDuration time.Duration
}

func NewDeferMessage

func NewDeferMessage(topic string, deferDuration time.Duration, body []byte) *Message

func NewMessage

func NewMessage(topic string, body []byte) *Message

func NewOrderlyMessage

func NewOrderlyMessage(topic, partitionKey string, body []byte) *Message

func (*Message) WithDeferDuration

func (m *Message) WithDeferDuration(deferDuration time.Duration) *Message

func (*Message) WithPartitionKey

func (m *Message) WithPartitionKey(partitionKey string) *Message

func (*Message) WithProperties

func (m *Message) WithProperties(properties map[string]string) *Message

func (*Message) WithTag

func (m *Message) WithTag(tag string) *Message

type MessageExt

type MessageExt struct {
	Message
	MsgID string
}

type ProducerConfig

type ProducerConfig struct {
	// Name server address
	Addr []string
	// Timeout for producing one message
	ProduceTimeout time.Duration
	// Retry times for producing
	RetryTimes int
	// Use compression, default is no compression
	Compression CompressionCodec
	// Producer group name
	ProducerGroup *string
}

type SendResponse

type SendResponse struct {
	MessageID string
	Offset    int64
}