producer

Documentation

Overview

Package producer is the generic producer-side processing logic for the four modes direct, topic, fanout, headers.

Index

• Variables
• type Exchange
  • func NewDirectExchange(exchangeName string, routingKey string) *Exchange
  • func NewFanOutExchange(exchangeName string) *Exchange
  • func NewHeaderExchange(exchangeName string, headersType HeadersType, kv map[string]interface{}) *Exchange
  • func NewTopicExchange(exchangeName string, routingKey string) *Exchange
• type ExchangeDeclareOption
  • func WithExchangeDeclareArgs(args map[string]interface{}) ExchangeDeclareOption
  • func WithExchangeDeclareAutoDelete(enable bool) ExchangeDeclareOption
  • func WithExchangeDeclareDurable(enable bool) ExchangeDeclareOption
  • func WithExchangeDeclareInternal(enable bool) ExchangeDeclareOption
  • func WithExchangeDeclareNoWait(enable bool) ExchangeDeclareOption
• type HeadersType
• type Queue
  • func NewQueue(queueName string, conn *amqp.Connection, exchange *Exchange, ...) (*Queue, error)
  • func (q *Queue) Close()
  • func (q *Queue) Publish(ctx context.Context, body []byte) error
  • func (q *Queue) PublishHeaders(ctx context.Context, headersKey map[string]interface{}, body []byte) error
  • func (q *Queue) PublishTopic(ctx context.Context, topicKey string, body []byte) error
• type QueueBindOption
  • func WithQueueBindArgs(args map[string]interface{}) QueueBindOption
  • func WithQueueBindNoWait(enable bool) QueueBindOption
• type QueueDeclareOption
  • func WithQueueDeclareArgs(args map[string]interface{}) QueueDeclareOption
  • func WithQueueDeclareAutoDelete(enable bool) QueueDeclareOption
  • func WithQueueDeclareDurable(enable bool) QueueDeclareOption
  • func WithQueueDeclareExclusive(enable bool) QueueDeclareOption
  • func WithQueueDeclareNoWait(enable bool) QueueDeclareOption
• type QueueOption
  • func WithExchangeDeclareOptions(opts ...ExchangeDeclareOption) QueueOption
  • func WithQueueBindOptions(opts ...QueueBindOption) QueueOption
  • func WithQueueDeclareOptions(opts ...QueueDeclareOption) QueueOption
  • func WithQueuePublishImmediate(enable bool) QueueOption
  • func WithQueuePublishMandatory(enable bool) QueueOption

Variables

var ErrClosed = amqp.ErrClosed

ErrClosed closed

Types

type Exchange

type Exchange struct {
	Headers map[string]interface{} // this field is required if eType=headers.
	// contains filtered or unexported fields
}

Exchange rabbitmq minimum management unit

func NewDirectExchange

func NewDirectExchange(exchangeName string, routingKey string) *Exchange

NewDirectExchange create a direct exchange

func NewFanOutExchange

func NewFanOutExchange(exchangeName string) *Exchange

NewFanOutExchange create a fanout exchange

func NewHeaderExchange

func NewHeaderExchange(exchangeName string, headersType HeadersType, kv map[string]interface{}) *Exchange

NewHeaderExchange create a headers exchange, the headerType supports "all" and "any"

func NewTopicExchange

func NewTopicExchange(exchangeName string, routingKey string) *Exchange

NewTopicExchange create a topic exchange

type ExchangeDeclareOption

type ExchangeDeclareOption func(*exchangeDeclareOptions)

ExchangeDeclareOption declare exchange option.

func WithExchangeDeclareArgs

func WithExchangeDeclareArgs(args map[string]interface{}) ExchangeDeclareOption

WithExchangeDeclareArgs set exchange declare args option.

func WithExchangeDeclareAutoDelete

func WithExchangeDeclareAutoDelete(enable bool) ExchangeDeclareOption

WithExchangeDeclareAutoDelete set exchange declare auto delete option.

func WithExchangeDeclareDurable

func WithExchangeDeclareDurable(enable bool) ExchangeDeclareOption

WithExchangeDeclareDurable set exchange declare durable option.

func WithExchangeDeclareInternal

func WithExchangeDeclareInternal(enable bool) ExchangeDeclareOption

WithExchangeDeclareInternal set exchange declare internal option.

func WithExchangeDeclareNoWait

func WithExchangeDeclareNoWait(enable bool) ExchangeDeclareOption

WithExchangeDeclareNoWait set exchange declare no wait option.

type HeadersType

type HeadersType = string

HeadersType headers type

const (

	// HeadersTypeAll all
	HeadersTypeAll HeadersType = "all"
	// HeadersTypeAny any
	HeadersTypeAny HeadersType = "any"
)

type Queue

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

Queue session

func NewQueue

func NewQueue(queueName string, conn *amqp.Connection, exchange *Exchange, opts ...QueueOption) (*Queue, error)

NewQueue create a queue

func (*Queue) Close

func (q *Queue) Close()

Close the queue

func (*Queue) Publish

func (q *Queue) Publish(ctx context.Context, body []byte) error

Publish send direct or fanout type message

func (*Queue) PublishHeaders

func (q *Queue) PublishHeaders(ctx context.Context, headersKey map[string]interface{}, body []byte) error

PublishHeaders send headers type message

func (*Queue) PublishTopic

func (q *Queue) PublishTopic(ctx context.Context, topicKey string, body []byte) error

PublishTopic send topic type message

type QueueBindOption

type QueueBindOption func(*queueBindOptions)

QueueBindOption declare queue bind option.

func WithQueueBindArgs

func WithQueueBindArgs(args map[string]interface{}) QueueBindOption

WithQueueBindArgs set queue bind args option.

func WithQueueBindNoWait

func WithQueueBindNoWait(enable bool) QueueBindOption

WithQueueBindNoWait set queue bind no wait option.

type QueueDeclareOption

type QueueDeclareOption func(*queueDeclareOptions)

QueueDeclareOption declare queue option.

func WithQueueDeclareArgs

func WithQueueDeclareArgs(args map[string]interface{}) QueueDeclareOption

WithQueueDeclareArgs set queue declare args option.

func WithQueueDeclareAutoDelete

func WithQueueDeclareAutoDelete(enable bool) QueueDeclareOption

WithQueueDeclareAutoDelete set queue declare auto delete option.

func WithQueueDeclareDurable

func WithQueueDeclareDurable(enable bool) QueueDeclareOption

WithQueueDeclareDurable set queue declare durable option.

func WithQueueDeclareExclusive

func WithQueueDeclareExclusive(enable bool) QueueDeclareOption

WithQueueDeclareExclusive set queue declare exclusive option.

func WithQueueDeclareNoWait

func WithQueueDeclareNoWait(enable bool) QueueDeclareOption

WithQueueDeclareNoWait set queue declare no wait option.

type QueueOption

type QueueOption func(*queueOptions)

QueueOption queue option.

func WithExchangeDeclareOptions

func WithExchangeDeclareOptions(opts ...ExchangeDeclareOption) QueueOption

WithExchangeDeclareOptions set exchange declare option.

func WithQueueBindOptions

func WithQueueBindOptions(opts ...QueueBindOption) QueueOption

WithQueueBindOptions set queue bind option.

func WithQueueDeclareOptions

func WithQueueDeclareOptions(opts ...QueueDeclareOption) QueueOption

WithQueueDeclareOptions set queue declare option.

func WithQueuePublishImmediate

func WithQueuePublishImmediate(enable bool) QueueOption

WithQueuePublishImmediate set queue publish immediate option.

func WithQueuePublishMandatory

func WithQueuePublishMandatory(enable bool) QueueOption

WithQueuePublishMandatory set queue publish mandatory option.