producer

Documentation

Overview

*

• builtin interceptor

Index

• func MarshalMessageBatch(msgs ...*primitive.Message) []byte
• func NewDefaultProducer(opts ...Option) (*defaultProducer, error)
• func NewTransactionProducer(listener primitive.TransactionListener, opts ...Option) (*transactionProducer, error)
• type Option
  • func WithCompressLevel(level int) Option
  • func WithCompressMsgBodyOverHowmuch(threshold int) Option
  • func WithCreateTopicKey(topic string) Option
  • func WithCredentials(c primitive.Credentials) Option
  • func WithDefaultTopicQueueNums(queueNum int) Option
  • func WithGroupName(group string) Option
  • func WithInstanceName(name string) Option
  • func WithInterceptor(f ...primitive.Interceptor) Option
  • func WithNameServer(nameServers primitive.NamesrvAddr) Option
  • func WithNameServerDomain(nameServerUrl string) Option
  • func WithNamespace(namespace string) Option
  • func WithNsResolver(resolver primitive.NsResolver) Option
  • func WithQueueSelector(s QueueSelector) Option
  • func WithRetry(retries int) Option
  • func WithSendMsgTimeout(duration time.Duration) Option
  • func WithTrace(traceCfg *primitive.TraceConfig) Option
  • func WithUnitName(unitName string) Option
  • func WithVIPChannel(enable bool) Option
• type QueueSelector
  • func NewHashQueueSelector() QueueSelector
  • func NewManualQueueSelector() QueueSelector
  • func NewRandomQueueSelector() QueueSelector
  • func NewRoundRobinQueueSelector() QueueSelector

Functions

func MarshalMessageBatch

func MarshalMessageBatch(msgs ...*primitive.Message) []byte

func NewDefaultProducer

func NewDefaultProducer(opts ...Option) (*defaultProducer, error)

func NewTransactionProducer

func NewTransactionProducer(listener primitive.TransactionListener, opts ...Option) (*transactionProducer, error)

TODO: checkLocalTransaction

Types

type Option

type Option func(*producerOptions)

func WithCompressLevel

func WithCompressLevel(level int) Option

WithCompressLevel set compress level (0~9) 0 stands for best speed 9 stands for best compression ratio

func WithCompressMsgBodyOverHowmuch

func WithCompressMsgBodyOverHowmuch(threshold int) Option

WithCompressMsgBodyOverHowmuch set compression threshold

func WithCreateTopicKey

func WithCreateTopicKey(topic string) Option

func WithCredentials

func WithCredentials(c primitive.Credentials) Option

func WithDefaultTopicQueueNums

func WithDefaultTopicQueueNums(queueNum int) Option

func WithGroupName

func WithGroupName(group string) Option

WithGroupName set group name address

func WithInstanceName

func WithInstanceName(name string) Option

func WithInterceptor

func WithInterceptor(f ...primitive.Interceptor) Option

func WithNameServer

func WithNameServer(nameServers primitive.NamesrvAddr) Option

WithNameServer set NameServer address, only support one NameServer cluster in alpha2

func WithNameServerDomain

func WithNameServerDomain(nameServerUrl string) Option

WithNameServerDomain set NameServer domain

func WithNamespace

func WithNamespace(namespace string) Option

WithNamespace set the namespace of producer

func WithNsResolver

func WithNsResolver(resolver primitive.NsResolver) Option

WithNsResolver set nameserver resolver to fetch nameserver addr

func WithQueueSelector

func WithQueueSelector(s QueueSelector) Option

func WithRetry

func WithRetry(retries int) Option

WithRetry return a Option that specifies the retry times when send failed. TODO: use retry middleware instead

func WithSendMsgTimeout

func WithSendMsgTimeout(duration time.Duration) Option

func WithTrace

func WithTrace(traceCfg *primitive.TraceConfig) Option

WithTrace support rocketmq trace: https://github.com/apache/rocketmq/wiki/RIP-6-Message-Trace.

func WithUnitName

func WithUnitName(unitName string) Option

WithUnitName set the name of specified unit

func WithVIPChannel

func WithVIPChannel(enable bool) Option

type QueueSelector

type QueueSelector interface {
	Select(msg *primitive.Message, mqs []*primitive.MessageQueue, lastBrokerName string) *primitive.MessageQueue
}

func NewHashQueueSelector

func NewHashQueueSelector() QueueSelector

func NewManualQueueSelector

func NewManualQueueSelector() QueueSelector

func NewRandomQueueSelector

func NewRandomQueueSelector() QueueSelector

func NewRoundRobinQueueSelector

func NewRoundRobinQueueSelector() QueueSelector