amqp

Documentation

Index

• type Consumer
  • func NewConsumer(opts ...Option) *Consumer
  • func (s *Consumer) Start(ctx context.Context) error
  • func (s *Consumer) Stop(ctx context.Context) error
• type ExchangeOption
• type MessageHandle
• type Option
  • func WithDsn(s string) Option
  • func WithExchangeDeclare(es ...ExchangeOption) Option
  • func WithName(s string) Option
  • func WithOnError(f func(error)) Option
  • func WithOnSubMessage(queue string, f MessageHandle) Option
  • func WithQos(prefetchCount, prefetchSize int, global bool) Option
  • func WithQueueDeclare(qs ...QueueOption) Option
• type Producer
  • func NewProducer(ctx context.Context, opts ...Option) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) Publish(ctx context.Context, exchange, key string, data []byte) error
  • func (p *Producer) PublishMsg(ctx context.Context, exchange, key string, msg amqp091.Publishing) error
• type QueueBind
• type QueueOption

Types

type Consumer

type Consumer struct {
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(opts ...Option) *Consumer

func (*Consumer) Start

func (s *Consumer) Start(ctx context.Context) error

func (*Consumer) Stop

func (s *Consumer) Stop(ctx context.Context) error

type ExchangeOption

type ExchangeOption struct {
	// 交换机名称
	Name string
	// 交换机类型 如 fanout direct topic
	Kind string
	// Durable 持久化
	Durable bool
	// AutoDelete设置为 true 表示自动删除 慎用 不是自动删除交换机。
	AutoDelete bool
}

ExchangeOption 交换机信息 sub & pub 都可以用

type MessageHandle

type MessageHandle func(ctx context.Context, msg amqp091.Delivery)

MessageHandle ding

type Option

type Option func(*option)

func WithDsn

func WithDsn(s string) Option

func WithExchangeDeclare

func WithExchangeDeclare(es ...ExchangeOption) Option

func WithName

func WithName(s string) Option

func WithOnError

func WithOnError(f func(error)) Option

func WithOnSubMessage

func WithOnSubMessage(queue string, f MessageHandle) Option

WithOnSubMessage only sub

func WithQos

func WithQos(prefetchCount, prefetchSize int, global bool) Option

func WithQueueDeclare

func WithQueueDeclare(qs ...QueueOption) Option

type Producer

type Producer struct {
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(ctx context.Context, opts ...Option) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, exchange, key string, data []byte) error

func (*Producer) PublishMsg

func (p *Producer) PublishMsg(ctx context.Context, exchange, key string, msg amqp091.Publishing) error

type QueueBind

type QueueBind struct {
	Exchange   string
	RoutingKey string
}

type QueueOption

type QueueOption struct {
	Queue      string
	Qos        int
	Durable    bool
	AutoDelete bool
	// 如果配置则自动进行交换机绑定
	BindExchange []QueueBind
}

SubQueueOption sub 声明队列配置