producer

Documentation

Overview

*

• builtin interceptor

Index

• Variables
• func MarshalMessageBatch(msgs ...*primitive.Message) []byte
• func NewDefaultProducer(opts ...Option) (*defaultProducer, error)
• func NewTransactionProducer(listener primitive.TransactionListener, opts ...Option) (*transactionProducer, error)
• type 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 WithNsResovler(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 WithVIPChannel(enable bool) Option
• type QueueSelector
  • func NewHashQueueSelector() QueueSelector
  • func NewManualQueueSelector() QueueSelector
  • func NewRandomQueueSelector() QueueSelector
  • func NewRoundRobinQueueSelector() QueueSelector

Variables

var (
	ErrTopicEmpty   = errors.New("topic is nil")
	ErrMessageEmpty = errors.New("message is nil")
	ErrNotRunning   = errors.New("producer not started")
)

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 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 WithNsResovler

func WithNsResovler(resolver primitive.NsResolver) Option

WithNsResovler 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 WithVIPChannel

func WithVIPChannel(enable bool) Option

type QueueSelector

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

func NewHashQueueSelector

func NewHashQueueSelector() QueueSelector

func NewManualQueueSelector

func NewManualQueueSelector() QueueSelector

func NewRandomQueueSelector

func NewRandomQueueSelector() QueueSelector

func NewRoundRobinQueueSelector

func NewRoundRobinQueueSelector() QueueSelector