rmqproducerx

Documentation

Index

• func Connect(configKey string, c *Config)
• func GetClient(ctx context.Context, keys ...string) rocketmq.Producer
• func Init(cm components.ConfigMap[*Config]) (func() error, error)
• func IsHealth() (err error)
• type Config
• type ProduceOption
  • func WithDelayTimeLevel(level int) ProduceOption
  • func WithKeys(keys ...string) ProduceOption
  • func WithProperties(properties map[string]string) ProduceOption
  • func WithProperty(key string, value string) ProduceOption
  • func WithShardingKey(key string) ProduceOption
  • func WithTag(tag string) ProduceOption
• type Producer
  • func GetProducer[T any](ctx context.Context, topic string, keys ...string) *Producer[T]
  • func (p *Producer[T]) GetInstance() rocketmq.Producer
  • func (p *Producer[T]) Request(ctx context.Context, message T, ttl time.Duration, options ...ProduceOption) (*primitive.Message, error)
  • func (p *Producer[T]) RequestAsync(ctx context.Context, ...) error
  • func (p *Producer[T]) SendAsync(ctx context.Context, message T, options ...ProduceOption) error
  • func (p *Producer[T]) SendAsyncWithCallback(ctx context.Context, message T, ...) error
  • func (p *Producer[T]) SendOneWay(ctx context.Context, message T, options ...ProduceOption) error
  • func (p *Producer[T]) SendSync(ctx context.Context, message T, options ...ProduceOption) (*primitive.SendResult, error)

Functions

func Connect

func Connect(configKey string, c *Config)

func GetClient

func GetClient(ctx context.Context, keys ...string) rocketmq.Producer

func Init

func Init(cm components.ConfigMap[*Config]) (func() error, error)

func IsHealth

func IsHealth() (err error)

Types

type Config

type Config struct {
	NameServer string `json:"name_server"`
}

type ProduceOption

type ProduceOption func(*primitive.Message) *primitive.Message

func WithDelayTimeLevel

func WithDelayTimeLevel(level int) ProduceOption

func WithKeys

func WithKeys(keys ...string) ProduceOption

func WithProperties

func WithProperties(properties map[string]string) ProduceOption

func WithProperty

func WithProperty(key string, value string) ProduceOption

func WithShardingKey

func WithShardingKey(key string) ProduceOption

func WithTag

func WithTag(tag string) ProduceOption

type Producer

type Producer[T any] struct {
	// contains filtered or unexported fields
}

func GetProducer

func GetProducer[T any](ctx context.Context, topic string, keys ...string) *Producer[T]

func (*Producer[T]) GetInstance

func (p *Producer[T]) GetInstance() rocketmq.Producer

func (*Producer[T]) Request

func (p *Producer[T]) Request(ctx context.Context, message T, ttl time.Duration, options ...ProduceOption) (*primitive.Message, error)

func (*Producer[T]) RequestAsync

func (p *Producer[T]) RequestAsync(ctx context.Context, callback func(ctx context.Context, msg *primitive.Message, err error), message T, ttl time.Duration, options ...ProduceOption) error

func (*Producer[T]) SendAsync

func (p *Producer[T]) SendAsync(ctx context.Context, message T, options ...ProduceOption) error

func (*Producer[T]) SendAsyncWithCallback

func (p *Producer[T]) SendAsyncWithCallback(ctx context.Context, message T, callback func(ctx context.Context, result *primitive.SendResult, err error), options ...ProduceOption) error

func (*Producer[T]) SendOneWay

func (p *Producer[T]) SendOneWay(ctx context.Context, message T, options ...ProduceOption) error

func (*Producer[T]) SendSync

func (p *Producer[T]) SendSync(ctx context.Context, message T, options ...ProduceOption) (*primitive.SendResult, error)