rabbitmq

package module
v0.15.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 27, 2024 License: MIT Imports: 12 Imported by: 121

README

go-rabbitmq

A wrapper of rabbitmq/amqp091-go that provides reconnection logic and sane defaults. Hit the project with a star if you find it useful ⭐

Supported by Boot.dev. If you'd like to learn about RabbitMQ and Go, you can check out my course here.

Deploy

Motivation

Streadway's AMQP library is currently the most robust and well-supported Go client I'm aware of. It's a fantastic option and I recommend starting there and seeing if it fulfills your needs. Their project has made an effort to stay within the scope of the AMQP protocol, as such, no reconnection logic and few ease-of-use abstractions are provided.

Goal

The goal with go-rabbitmq is to provide most (but not all) of the nitty-gritty functionality of Streadway's AMQP, but to make it easier to work with via a higher-level API. go-rabbitmq is also built specifically for Rabbit, not for the AMQP protocol. In particular, we want:

  • Automatic reconnection
  • Multithreaded consumers via a handler function
  • Reasonable defaults
  • Flow control handling
  • TCP block handling

⚙️ Installation

Inside a Go module:

go get github.com/wagslane/go-rabbitmq

🚀 Quick Start Consumer

Take note of the optional options parameters after the queue name. The queue will be declared automatically, but the exchange will not. You'll also probably want to bind to at least one routing key.

conn, err := rabbitmq.NewConn(
	"amqp://guest:guest@localhost",
	rabbitmq.WithConnectionOptionsLogging,
)
if err != nil {
	log.Fatal(err)
}
defer conn.Close()

consumer, err := rabbitmq.NewConsumer(
	conn,
	"my_queue",
	rabbitmq.WithConsumerOptionsRoutingKey("my_routing_key"),
	rabbitmq.WithConsumerOptionsExchangeName("events"),
	rabbitmq.WithConsumerOptionsExchangeDeclare,
)
if err != nil {
	log.Fatal(err)
}
defer consumer.Close()

err = consumer.Run(func(d rabbitmq.Delivery) rabbitmq.Action {
	log.Printf("consumed: %v", string(d.Body))
	// rabbitmq.Ack, rabbitmq.NackDiscard, rabbitmq.NackRequeue
	return rabbitmq.Ack
})
if err != nil {
	log.Fatal(err)
}

🚀 Quick Start Publisher

The exchange is not declared by default, that's why I recommend using the following options.

conn, err := rabbitmq.NewConn(
	"amqp://guest:guest@localhost",
	rabbitmq.WithConnectionOptionsLogging,
)
if err != nil {
	log.Fatal(err)
}
defer conn.Close()

publisher, err := rabbitmq.NewPublisher(
	conn,
	rabbitmq.WithPublisherOptionsLogging,
	rabbitmq.WithPublisherOptionsExchangeName("events"),
	rabbitmq.WithPublisherOptionsExchangeDeclare,
)
if err != nil {
	log.Fatal(err)
}
defer publisher.Close()

err = publisher.Publish(
	[]byte("hello, world"),
	[]string{"my_routing_key"},
	rabbitmq.WithPublishOptionsContentType("application/json"),
	rabbitmq.WithPublishOptionsExchange("events"),
)
if err != nil {
	log.Println(err)
}

Other usage examples

See the examples directory for more ideas.

Options and configuring

  • By default, queues are declared if they didn't already exist by new consumers
  • By default, routing-key bindings are declared by consumers if you're using WithConsumerOptionsRoutingKey
  • By default, exchanges are not declared by publishers or consumers if they don't already exist, hence WithPublisherOptionsExchangeDeclare and WithConsumerOptionsExchangeDeclare.

Read up on all the options in the GoDoc, there are quite a few of them. I try to pick sane and simple defaults.

Closing and resources

Close your publishers and consumers when you're done with them and do not attempt to reuse them. Only close the connection itself once you've closed all associated publishers and consumers.

Stability

Note that the API is currently in v0. I don't plan on huge changes, but there may be some small breaking changes before we hit v1.

Integration testing

By setting ENABLE_DOCKER_INTEGRATION_TESTS=TRUE during go test -v ./..., the integration tests will run. These launch a rabbitmq container in the local Docker daemon and test some publish/consume actions.

See integration_test.go.

💬 Contact

Twitter Follow

Submit an issue here on GitHub

Transient Dependencies

My goal is to keep dependencies limited to 1, github.com/rabbitmq/amqp091-go.

👏 Contributing

I would love your help! Contribute by forking the repo and opening pull requests. Please ensure that your code passes the existing tests and linting, and write tests to test your changes if applicable.

All pull requests should be submitted to the main branch.

Documentation

Index

Constants

View Source 
const (
	Transient  uint8 = amqp.Transient
	Persistent uint8 = amqp.Persistent
)

DeliveryMode. Transient means higher throughput but messages will not be restored on broker restart. The delivery mode of publishings is unrelated to the durability of the queues they reside on. Transient messages will not be restored to durable queues, persistent messages will be restored to durable queues and lost on non-durable queues during server restart.

This remains typed as uint8 to match Publishing.DeliveryMode. Other delivery modes specific to custom queue implementations are not enumerated here.

Variables

This section is empty.

Functions

func WithConnectionOptionsConfig added in v0.11.0 

func WithConnectionOptionsConfig(cfg Config) func(options *ConnectionOptions)

WithConnectionOptionsConfig sets the Config used in the connection

func WithConnectionOptionsLogger added in v0.11.0 

func WithConnectionOptionsLogger(log Logger) func(options *ConnectionOptions)

WithConnectionOptionsLogger sets logging to true on the consumer options and sets the

func WithConnectionOptionsLogging added in v0.11.0 

func WithConnectionOptionsLogging(options *ConnectionOptions)

WithConnectionOptionsLogging sets logging to true on the consumer options and sets the

func WithConnectionOptionsReconnectInterval added in v0.11.0 

func WithConnectionOptionsReconnectInterval(interval time.Duration) func(options *ConnectionOptions)

WithConnectionOptionsReconnectInterval sets the reconnection interval

func WithConsumerOptionsBinding added in v0.11.0 

func WithConsumerOptionsBinding(binding Binding) func(*ConsumerOptions)

WithConsumerOptionsBinding adds a new binding to the queue which allows you to set the binding options on a per-binding basis. Keep in mind that everything in the BindingOptions struct will default to the zero value. If you want to declare your bindings for example, be sure to set Declare=true

func WithConsumerOptionsConcurrency added in v0.11.0 

func WithConsumerOptionsConcurrency(concurrency int) func(*ConsumerOptions)

WithConsumerOptionsConcurrency returns a function that sets the concurrency, which means that many goroutines will be spawned to run the provided handler on messages

func WithConsumerOptionsConsumerAutoAck added in v0.11.0 

func WithConsumerOptionsConsumerAutoAck(autoAck bool) func(*ConsumerOptions)

WithConsumerOptionsConsumerAutoAck returns a function that sets the auto acknowledge property on the server of this consumer if unset the default will be used (false)

func WithConsumerOptionsConsumerExclusive added in v0.11.0 

func WithConsumerOptionsConsumerExclusive(options *ConsumerOptions)

WithConsumerOptionsConsumerExclusive sets the consumer to exclusive, which means the server will ensure that this is the sole consumer from this queue. When exclusive is false, the server will fairly distribute deliveries across multiple consumers.

func WithConsumerOptionsConsumerName added in v0.11.0 

func WithConsumerOptionsConsumerName(consumerName string) func(*ConsumerOptions)

WithConsumerOptionsConsumerName returns a function that sets the name on the server of this consumer if unset a random name will be given

func WithConsumerOptionsConsumerNoWait added in v0.11.0 

func WithConsumerOptionsConsumerNoWait(options *ConsumerOptions)

WithConsumerOptionsConsumerNoWait sets the consumer to nowait, which means it does not wait for the server to confirm the request and immediately begin deliveries. If it is not possible to consume, a channel exception will be raised and the channel will be closed.

func WithConsumerOptionsExchangeArgs added in v0.11.0 

func WithConsumerOptionsExchangeArgs(args Table) func(*ConsumerOptions)

WithConsumerOptionsExchangeArgs adds optional args to the exchange

func WithConsumerOptionsExchangeAutoDelete added in v0.11.0 

func WithConsumerOptionsExchangeAutoDelete(options *ConsumerOptions)

WithConsumerOptionsExchangeAutoDelete ensures the exchange is an auto-delete exchange

func WithConsumerOptionsExchangeDeclare added in v0.11.0 

func WithConsumerOptionsExchangeDeclare(options *ConsumerOptions)

WithConsumerOptionsExchangeDeclare stops this library from declaring the exchanges existance

func WithConsumerOptionsExchangeDurable added in v0.11.0 

func WithConsumerOptionsExchangeDurable(options *ConsumerOptions)

WithConsumerOptionsExchangeDurable ensures the exchange is a durable exchange

func WithConsumerOptionsExchangeInternal added in v0.11.0 

func WithConsumerOptionsExchangeInternal(options *ConsumerOptions)

WithConsumerOptionsExchangeInternal ensures the exchange is an internal exchange

func WithConsumerOptionsExchangeKind added in v0.11.0 

func WithConsumerOptionsExchangeKind(kind string) func(*ConsumerOptions)

WithConsumerOptionsExchangeKind ensures the queue is a durable queue

func WithConsumerOptionsExchangeName added in v0.11.0 

func WithConsumerOptionsExchangeName(name string) func(*ConsumerOptions)

WithConsumerOptionsExchangeName sets the exchange name

func WithConsumerOptionsExchangeNoWait added in v0.11.0 

func WithConsumerOptionsExchangeNoWait(options *ConsumerOptions)

WithConsumerOptionsExchangeNoWait ensures the exchange is a no-wait exchange

func WithConsumerOptionsExchangeOptions added in v0.13.0 

func WithConsumerOptionsExchangeOptions(exchangeOptions ExchangeOptions) func(*ConsumerOptions)

WithConsumerOptionsExchangeOptions adds a new exchange to the consumer, this should probably only be used if you want to to consume from multiple exchanges on the same consumer

func WithConsumerOptionsExchangePassive added in v0.11.0 

func WithConsumerOptionsExchangePassive(options *ConsumerOptions)

WithConsumerOptionsExchangePassive ensures the exchange is a passive exchange

func WithConsumerOptionsForceShutdown added in v0.14.0 

func WithConsumerOptionsForceShutdown(options *ConsumerOptions)

WithConsumerOptionsForceShutdown tells the consumer to not wait for the handler to complete in consumer.Close

func WithConsumerOptionsLogger added in v0.5.0 

func WithConsumerOptionsLogger(log logger.Logger) func(options *ConsumerOptions)

WithConsumerOptionsLogger sets logging to a custom interface. Use WithConsumerOptionsLogging to just log to stdout.

func WithConsumerOptionsLogging added in v0.3.0 

func WithConsumerOptionsLogging(options *ConsumerOptions)

WithConsumerOptionsLogging uses a default logger that writes to std out

func WithConsumerOptionsQOSGlobal added in v0.11.0 

func WithConsumerOptionsQOSGlobal(options *ConsumerOptions)

WithConsumerOptionsQOSGlobal sets the qos on the channel to global, which means these QOS settings apply to ALL existing and future consumers on all channels on the same connection

func WithConsumerOptionsQOSPrefetch added in v0.11.0 

func WithConsumerOptionsQOSPrefetch(prefetchCount int) func(*ConsumerOptions)

WithConsumerOptionsQOSPrefetch returns a function that sets the prefetch count, which means that many messages will be fetched from the server in advance to help with throughput. This doesn't affect the handler, messages are still processed one at a time.

func WithConsumerOptionsQueueArgs added in v0.11.0 

func WithConsumerOptionsQueueArgs(args Table) func(*ConsumerOptions)

WithConsumerOptionsQueueArgs adds optional args to the queue

func WithConsumerOptionsQueueAutoDelete added in v0.11.0 

func WithConsumerOptionsQueueAutoDelete(options *ConsumerOptions)

WithConsumerOptionsQueueAutoDelete ensures the queue is an auto-delete queue

func WithConsumerOptionsQueueDurable added in v0.11.0 

func WithConsumerOptionsQueueDurable(options *ConsumerOptions)

WithConsumerOptionsQueueDurable ensures the queue is a durable queue

func WithConsumerOptionsQueueExclusive added in v0.11.0 

func WithConsumerOptionsQueueExclusive(options *ConsumerOptions)

WithConsumerOptionsQueueExclusive ensures the queue is an exclusive queue

func WithConsumerOptionsQueueNoDeclare added in v0.11.0 

func WithConsumerOptionsQueueNoDeclare(options *ConsumerOptions)

WithConsumerOptionsQueueNoDeclare will turn off the declaration of the queue's existance upon startup

func WithConsumerOptionsQueueNoWait added in v0.11.0 

func WithConsumerOptionsQueueNoWait(options *ConsumerOptions)

WithConsumerOptionsQueueNoWait ensures the queue is a no-wait queue

func WithConsumerOptionsQueuePassive added in v0.11.0 

func WithConsumerOptionsQueuePassive(options *ConsumerOptions)

WithConsumerOptionsQueuePassive ensures the queue is a passive queue

func WithConsumerOptionsQueueQuorum added in v0.11.1 

func WithConsumerOptionsQueueQuorum(options *ConsumerOptions)

WithConsumerOptionsQueueQuorum sets the queue a quorum type, which means multiple nodes in the cluster will have the messages distributed amongst them for higher reliability

func WithConsumerOptionsRoutingKey added in v0.11.0 

func WithConsumerOptionsRoutingKey(routingKey string) func(*ConsumerOptions)

WithConsumerOptionsRoutingKey binds the queue to a routing key with the default binding options

func WithPublishOptionsAppID added in v0.7.0 

func WithPublishOptionsAppID(appID string) func(*PublishOptions)

WithPublishOptionsAppID returns a function that sets the application id

func WithPublishOptionsContentEncoding added in v0.7.0 

func WithPublishOptionsContentEncoding(contentEncoding string) func(*PublishOptions)

WithPublishOptionsContentEncoding returns a function that sets the content encoding, i.e. "utf-8"

func WithPublishOptionsContentType added in v0.3.0 

func WithPublishOptionsContentType(contentType string) func(*PublishOptions)

WithPublishOptionsContentType returns a function that sets the content type, i.e. "application/json"

func WithPublishOptionsCorrelationID added in v0.7.0 

func WithPublishOptionsCorrelationID(correlationID string) func(*PublishOptions)

WithPublishOptionsCorrelationID returns a function that sets the content correlation identifier

func WithPublishOptionsExchange added in v0.3.0 

func WithPublishOptionsExchange(exchange string) func(*PublishOptions)

WithPublishOptionsExchange returns a function that sets the exchange to publish to

func WithPublishOptionsExpiration added in v0.6.0 

func WithPublishOptionsExpiration(expiration string) func(options *PublishOptions)

WithPublishOptionsExpiration returns a function that sets the expiry/TTL of a message. As per RabbitMq spec, it must be a string value in milliseconds.

func WithPublishOptionsHeaders added in v0.6.0 

func WithPublishOptionsHeaders(headers Table) func(*PublishOptions)

WithPublishOptionsHeaders returns a function that sets message header values, i.e. "msg-id"

func WithPublishOptionsImmediate added in v0.3.0 

func WithPublishOptionsImmediate(options *PublishOptions)

WithPublishOptionsImmediate makes the publishing immediate, which means when a consumer is not available to immediately handle the new message, a message will be sent back on the returns channel for you to handle

func WithPublishOptionsMandatory added in v0.3.0 

func WithPublishOptionsMandatory(options *PublishOptions)

WithPublishOptionsMandatory makes the publishing mandatory, which means when a queue is not bound to the routing key a message will be sent back on the returns channel for you to handle

func WithPublishOptionsMessageID added in v0.7.0 

func WithPublishOptionsMessageID(messageID string) func(*PublishOptions)

WithPublishOptionsMessageID returns a function that sets the message identifier

func WithPublishOptionsPersistentDelivery added in v0.3.0 

func WithPublishOptionsPersistentDelivery(options *PublishOptions)

WithPublishOptionsPersistentDelivery sets the message to persist. Transient messages will not be restored to durable queues, persistent messages will be restored to durable queues and lost on non-durable queues during server restart. By default publishings are transient

func WithPublishOptionsPriority added in v0.7.0 

func WithPublishOptionsPriority(priority uint8) func(*PublishOptions)

WithPublishOptionsPriority returns a function that sets the content priority from 0 to 9

func WithPublishOptionsReplyTo added in v0.7.0 

func WithPublishOptionsReplyTo(replyTo string) func(*PublishOptions)

WithPublishOptionsReplyTo returns a function that sets the reply to field

func WithPublishOptionsTimestamp added in v0.7.0 

func WithPublishOptionsTimestamp(timestamp time.Time) func(*PublishOptions)

WithPublishOptionsTimestamp returns a function that sets the timestamp for the message

func WithPublishOptionsType added in v0.7.0 

func WithPublishOptionsType(messageType string) func(*PublishOptions)

WithPublishOptionsType returns a function that sets the message type name

func WithPublishOptionsUserID added in v0.7.0 

func WithPublishOptionsUserID(userID string) func(*PublishOptions)

WithPublishOptionsUserID returns a function that sets the user id i.e. "user"

func WithPublisherOptionsConfirm added in v0.12.4 

func WithPublisherOptionsConfirm(options *PublisherOptions)

WithPublisherOptionsConfirm enables confirm mode on the connection this is required if publisher confirmations should be used

func WithPublisherOptionsExchangeArgs added in v0.11.0 

func WithPublisherOptionsExchangeArgs(args Table) func(*PublisherOptions)

WithPublisherOptionsExchangeArgs adds optional args to the exchange

func WithPublisherOptionsExchangeAutoDelete added in v0.11.0 

func WithPublisherOptionsExchangeAutoDelete(options *PublisherOptions)

WithPublisherOptionsExchangeAutoDelete ensures the exchange is an auto-delete exchange

func WithPublisherOptionsExchangeDeclare added in v0.11.0 

func WithPublisherOptionsExchangeDeclare(options *PublisherOptions)

WithPublisherOptionsExchangeDeclare will create the exchange if it doesn't exist

func WithPublisherOptionsExchangeDurable added in v0.11.0 

func WithPublisherOptionsExchangeDurable(options *PublisherOptions)

WithPublisherOptionsExchangeDurable ensures the exchange is a durable exchange

func WithPublisherOptionsExchangeInternal added in v0.11.0 

func WithPublisherOptionsExchangeInternal(options *PublisherOptions)

WithPublisherOptionsExchangeInternal ensures the exchange is an internal exchange

func WithPublisherOptionsExchangeKind added in v0.11.0 

func WithPublisherOptionsExchangeKind(kind string) func(*PublisherOptions)

WithPublisherOptionsExchangeKind ensures the queue is a durable queue

func WithPublisherOptionsExchangeName added in v0.11.0 

func WithPublisherOptionsExchangeName(name string) func(*PublisherOptions)

WithPublisherOptionsExchangeName sets the exchange name

func WithPublisherOptionsExchangeNoWait added in v0.11.0 

func WithPublisherOptionsExchangeNoWait(options *PublisherOptions)

WithPublisherOptionsExchangeNoWait ensures the exchange is a no-wait exchange

func WithPublisherOptionsExchangePassive added in v0.11.0 

func WithPublisherOptionsExchangePassive(options *PublisherOptions)

WithPublisherOptionsExchangePassive ensures the exchange is a passive exchange

func WithPublisherOptionsLogger added in v0.5.0 

func WithPublisherOptionsLogger(log Logger) func(options *PublisherOptions)

WithPublisherOptionsLogger sets logging to a custom interface. Use WithPublisherOptionsLogging to just log to stdout.

func WithPublisherOptionsLogging added in v0.3.0 

func WithPublisherOptionsLogging(options *PublisherOptions)

WithPublisherOptionsLogging sets logging to true on the publisher options and sets the

Types

type Action added in v0.7.0 

type Action int

Action is an action that occurs after processed this delivery 

const (
	// Ack default ack this msg after you have successfully processed this delivery.
	Ack Action = iota
	// NackDiscard the message will be dropped or delivered to a server configured dead-letter queue.
	NackDiscard
	// NackRequeue deliver this message to a different consumer.
	NackRequeue
	// Message acknowledgement is left to the user using the msg.Ack() method
	Manual
)

type Binding added in v0.11.0 

type Binding struct {
	RoutingKey string
	BindingOptions
}

Binding describes the bhinding of a queue to a routing key on an exchange

type BindingOptions 

type BindingOptions struct {
	NoWait  bool
	Args    Table
	Declare bool
}

BindingOptions describes the options a binding can have

type Config added in v0.8.0 

type Config amqp.Config

Config wraps amqp.Config Config is used in DialConfig and Open to specify the desired tuning parameters used during a connection open handshake. The negotiated tuning will be stored in the returned connection's Config field.

type Confirmation added in v0.7.0 

type Confirmation struct {
	amqp.Confirmation
	ReconnectionCount int
}

Confirmation notifies the acknowledgment or negative acknowledgement of a publishing identified by its delivery tag. Use NotifyPublish to consume these events. ReconnectionCount is useful in that each time it increments, the DeliveryTag is reset to 0, meaning you can use ReconnectionCount+DeliveryTag to ensure uniqueness

type Conn added in v0.11.0 

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

Conn manages the connection to a rabbit cluster it is intended to be shared across publishers and consumers

func NewClusterConn added in v0.14.0 

func NewClusterConn(resolver Resolver, opts ...func(*ConnectionOptions)) (*Conn, error)

func NewConn added in v0.11.0 

func NewConn(url string, opts ...func(*ConnectionOptions)) (*Conn, error)

NewConn creates a new connection manager

func (*Conn) Close added in v0.11.0 

func (conn *Conn) Close() error

Close closes the connection, it's not safe for re-use. You should also close any consumers and publishers before closing the connection

type ConnectionOptions added in v0.11.0 

type ConnectionOptions struct {
	ReconnectInterval time.Duration
	Logger            Logger
	Config            Config
}

ConnectionOptions are used to describe how a new consumer will be created.

type Consumer 

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

Consumer allows you to create and connect to queues for data consumption.

func NewConsumer added in v0.2.0 

func NewConsumer(
	conn *Conn,
	queue string,
	optionFuncs ...func(*ConsumerOptions),
) (*Consumer, error)

NewConsumer returns a new Consumer connected to the given rabbitmq server.

func (*Consumer) Close added in v0.8.0 

func (consumer *Consumer) Close()

Close cleans up resources and closes the consumer. It waits for handler to finish before returning by default (use WithConsumerOptionsForceShutdown option to disable this behavior). Use CloseWithContext to specify a context to cancel the handler completion. It does not close the connection manager, just the subscription to the connection manager and the consuming goroutines. Only call once.

func (*Consumer) CloseWithContext added in v0.14.0 

func (consumer *Consumer) CloseWithContext(ctx context.Context)

CloseWithContext cleans up resources and closes the consumer. It waits for handler to finish before returning (use WithConsumerOptionsForceShutdown option to disable this behavior). Use the context to cancel the handler completion. CloseWithContext does not close the connection manager, just the subscription to the connection manager and the consuming goroutines. Only call once.

func (*Consumer) Run added in v0.13.0 

func (consumer *Consumer) Run(handler Handler) error

Run starts consuming with automatic reconnection handling. Do not reuse the consumer for anything other than to close it.

type ConsumerOptions 

type ConsumerOptions struct {
	RabbitConsumerOptions RabbitConsumerOptions
	QueueOptions          QueueOptions
	CloseGracefully       bool
	ExchangeOptions       []ExchangeOptions
	Concurrency           int
	Logger                logger.Logger
	QOSPrefetch           int
	QOSGlobal             bool
}

ConsumerOptions are used to describe how a new consumer will be created. If QueueOptions is not nil, the options will be used to declare a queue If ExchangeOptions is not nil, it will be used to declare an exchange If there are Bindings, the queue will be bound to them

type Delivery 

type Delivery struct {
	amqp.Delivery
}

Delivery captures the fields for a previously delivered message resident in a queue to be delivered by the server to a consumer from Channel.Consume or Channel.Get.

type ExchangeOptions added in v0.11.0 

type ExchangeOptions struct {
	Name       string
	Kind       string // possible values: empty string for default exchange or direct, topic, fanout
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Passive    bool // if false, a missing exchange will be created on the server
	Args       Table
	Declare    bool
	Bindings   []Binding
}

ExchangeOptions are used to configure an exchange. If the Passive flag is set the client will only check if the exchange exists on the server and that the settings match, no creation attempt will be made.

type Handler added in v0.7.0 

type Handler func(d Delivery) (action Action)

Handler defines the handler of each Delivery and return Action

type Logger added in v0.5.0 

type Logger logger.Logger

Logger is describes a logging structure. It can be set using WithPublisherOptionsLogger() or WithConsumerOptionsLogger().

type PublishOptions 

type PublishOptions struct {
	Exchange string
	// Mandatory fails to publish if there are no queues
	// bound to the routing key
	Mandatory bool
	// Immediate fails to publish if there are no consumers
	// that can ack bound to the queue on the routing key
	Immediate bool
	// MIME content type
	ContentType string
	// Transient (0 or 1) or Persistent (2)
	DeliveryMode uint8
	// Expiration time in ms that a message will expire from a queue.
	// See https://www.rabbitmq.com/ttl.html#per-message-ttl-in-publishers
	Expiration string
	// MIME content encoding
	ContentEncoding string
	// 0 to 9
	Priority uint8
	// correlation identifier
	CorrelationID string
	// address to to reply to (ex: RPC)
	ReplyTo string
	// message identifier
	MessageID string
	// message timestamp
	Timestamp time.Time
	// message type name
	Type string
	// creating user id - ex: "guest"
	UserID string
	// creating application id
	AppID string
	// Application or exchange specific fields,
	// the headers exchange will inspect this field.
	Headers Table
}

PublishOptions are used to control how data is published

type Publisher 

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

Publisher allows you to publish messages safely across an open connection

func NewPublisher added in v0.2.0 

func NewPublisher(conn *Conn, optionFuncs ...func(*PublisherOptions)) (*Publisher, error)

NewPublisher returns a new publisher with an open channel to the cluster. If you plan to enforce mandatory or immediate publishing, those failures will be reported on the channel of Returns that you should setup a listener on. Flow controls are automatically handled as they are sent from the server, and publishing will fail with an error when the server is requesting a slowdown

func (*Publisher) Close added in v0.8.0 

func (publisher *Publisher) Close()

Close closes the publisher and releases resources The publisher should be discarded as it's not safe for re-use Only call Close() once

func (*Publisher) NotifyPublish added in v0.7.0 

func (publisher *Publisher) NotifyPublish(handler func(p Confirmation))

NotifyPublish registers a listener for publish confirmations, must set ConfirmPublishings option These notifications are shared across an entire connection, so if you're creating multiple publishers on the same connection keep that in mind

func (*Publisher) NotifyReturn added in v0.7.0 

func (publisher *Publisher) NotifyReturn(handler func(r Return))

NotifyReturn registers a listener for basic.return methods. These can be sent from the server when a publish is undeliverable either from the mandatory or immediate flags. These notifications are shared across an entire connection, so if you're creating multiple publishers on the same connection keep that in mind

func (*Publisher) Publish 

func (publisher *Publisher) Publish(
	data []byte,
	routingKeys []string,
	optionFuncs ...func(*PublishOptions),
) error

Publish publishes the provided data to the given routing keys over the connection.

func (*Publisher) PublishWithContext added in v0.12.0 

func (publisher *Publisher) PublishWithContext(
	ctx context.Context,
	data []byte,
	routingKeys []string,
	optionFuncs ...func(*PublishOptions),
) error

PublishWithContext publishes the provided data to the given routing keys over the connection.

func (*Publisher) PublishWithDeferredConfirmWithContext added in v0.12.2 

func (publisher *Publisher) PublishWithDeferredConfirmWithContext(
	ctx context.Context,
	data []byte,
	routingKeys []string,
	optionFuncs ...func(*PublishOptions),
) (PublisherConfirmation, error)

PublishWithContext publishes the provided data to the given routing keys over the connection. if the publisher is in confirm mode (which can be either done by calling `NotifyPublish` with a custom handler or by using `WithPublisherOptionsConfirm`) a publisher confirmation is returned. This confirmation can be used to check if the message was actually published or wait for this to happen. If the publisher is not in confirm mode, the returned confirmation will always be nil.

type PublisherConfirmation added in v0.12.2 

type PublisherConfirmation []*amqp.DeferredConfirmation

type PublisherOptions added in v0.1.0 

type PublisherOptions struct {
	ExchangeOptions ExchangeOptions
	Logger          Logger
	ConfirmMode     bool
}

PublisherOptions are used to describe a publisher's configuration. Logger is a custom logging interface.

type QueueOptions 

type QueueOptions struct {
	Name       string
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Passive    bool // if false, a missing queue will be created on the server
	Args       Table
	Declare    bool
}

QueueOptions are used to configure a queue. A passive queue is assumed by RabbitMQ to already exist, and attempting to connect to a non-existent queue will cause RabbitMQ to throw an exception.

type RabbitConsumerOptions added in v0.11.0 

type RabbitConsumerOptions struct {
	Name      string
	AutoAck   bool
	Exclusive bool
	NoWait    bool
	NoLocal   bool
	Args      Table
}

RabbitConsumerOptions are used to configure the consumer on the rabbit server

type Resolver added in v0.14.0 

type Resolver = connectionmanager.Resolver

type Return 

type Return struct {
	amqp.Return
}

Return captures a flattened struct of fields returned by the server when a Publishing is unable to be delivered either due to the `mandatory` flag set and no route found, or `immediate` flag set and no free consumer.

type StaticResolver added in v0.14.0 

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

func NewStaticResolver added in v0.14.0 

func NewStaticResolver(urls []string, shuffle bool) *StaticResolver

func (*StaticResolver) Resolve added in v0.14.0 

func (r *StaticResolver) Resolve() ([]string, error)

type Table 

type Table map[string]interface{}

Table stores user supplied fields of the following types: 

bool
byte
float32
float64
int
int16
int32
int64
nil
string
time.Time
amqp.Decimal
amqp.Table
[]byte
[]interface{} - containing above types

Functions taking a table will immediately fail when the table contains a value of an unsupported type.

The caller must be specific in which precision of integer it wishes to encode.

Use a type assertion when reading values from a table for type conversion.

RabbitMQ expects int32 for integer values.

Source Files

View all Source files

Directories

Path Synopsis
examples
cluster
consumer
logger
multiconsumer
multipublisher
publisher
publisher_confirm
internal
channelmanager
connectionmanager
dispatcher
logger

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.