rabbitmq

package module
v0.6.0 Latest Latest
Warning

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

 Go to latest Published: Jun 6, 2021 License: MIT Imports: 6 Imported by: 147

README

go-rabbitmq

Wrapper of streadway/amqp that provides reconnection logic and sane defaults. Hit the project with a star if you find it useful ⭐

Supported by Qvault

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.

The goal with go-rabbitmq is to still provide most all of the nitty-gritty functionality of AMQP, but to make it easier to work with via a higher-level API. Particularly:

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

⚙️ Installation

Outside of a Go module:

go get github.com/wagslane/go-rabbitmq

🚀 Quick Start Consumer

Default options
consumer, err := rabbitmq.NewConsumer("amqp://user:pass@localhost")
if err != nil {
    log.Fatal(err)
}
err = consumer.StartConsuming(
    func(d rabbitmq.Delivery) bool {
        log.Printf("consumed: %v", string(d.Body))
        // true to ACK, false to NACK
        return true
    },
    "my_queue",
    []string{"routing_key1", "routing_key2"}
)
if err != nil {
    log.Fatal(err)
}
With options
consumer, err := rabbitmq.NewConsumer(
    "amqp://user:pass@localhost",
    rabbitmq.WithConsumerOptionsLogging,
)
if err != nil {
    log.Fatal(err)
}
err = consumer.StartConsuming(
    func(d rabbitmq.Delivery) bool {
        log.Printf("consumed: %v", string(d.Body))
        // true to ACK, false to NACK
        return true
    },
    "my_queue",
    []string{"routing_key1", "routing_key2"},
    rabbitmq.WithConsumeOptionsConcurrency(10),
    rabbitmq.WithConsumeOptionsQueueDurable,
    rabbitmq.WithConsumeOptionsQuorum,
)
if err != nil {
    log.Fatal(err)
}

🚀 Quick Start Publisher

Default options
publisher, returns, err := rabbitmq.NewPublisher("amqp://user:pass@localhost")
if err != nil {
    log.Fatal(err)
}
err = publisher.Publish([]byte("hello, world"), []string{"routing_key"})
if err != nil {
    log.Fatal(err)
}
With options
publisher, returns, err := rabbitmq.NewPublisher(
    "amqp://user:pass@localhost",
    // can pass nothing for no logging
    rabbitmq.WithPublisherOptionsLogging,
)
if err != nil {
    log.Fatal(err)
}
err = publisher.Publish(
    []byte("hello, world"),
    []string{"routing_key"},
    // leave blank for defaults
    rabbitmq.WithPublishOptionsContentType("application/json"),
    rabbitmq.WithPublishOptionsMandatory,
    rabbitmq.WithPublishOptionsPersistentDelivery,
)
if err != nil {
    log.Fatal(err)
}

go func() {
    for r := range returns {
        log.Printf("message returned from server: %s", string(r.Body))
    }
}()

💬 Contact

Twitter Follow

Submit an issue (above in the issues tab)

Transient Dependencies

My goal is to keep dependencies limited to 1, github.com/streadway/amqp.

👏 Contributing

I love 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 WithConsumeOptionsBindingExchangeArgs added in v0.6.0 

func WithConsumeOptionsBindingExchangeArgs(args Table) func(*ConsumeOptions)

WithConsumeOptionsBindingExchangeArgs returns a function that sets the binding exchange arguments that are specific to the server's implementation of the exchange

func WithConsumeOptionsBindingExchangeAutoDelete added in v0.6.0 

func WithConsumeOptionsBindingExchangeAutoDelete(options *ConsumeOptions)

WithConsumeOptionsBindingExchangeAutoDelete returns a function that sets the binding exchange autoDelete flag

func WithConsumeOptionsBindingExchangeDurable added in v0.6.0 

func WithConsumeOptionsBindingExchangeDurable(options *ConsumeOptions)

WithConsumeOptionsBindingExchangeDurable returns a function that sets the binding exchange durable flag

func WithConsumeOptionsBindingExchangeInternal added in v0.6.0 

func WithConsumeOptionsBindingExchangeInternal(options *ConsumeOptions)

WithConsumeOptionsBindingExchangeInternal returns a function that sets the binding exchange internal flag

func WithConsumeOptionsBindingExchangeKind added in v0.6.0 

func WithConsumeOptionsBindingExchangeKind(kind string) func(*ConsumeOptions)

WithConsumeOptionsBindingExchangeKind returns a function that sets the binding exchange kind/type

func WithConsumeOptionsBindingExchangeName added in v0.6.0 

func WithConsumeOptionsBindingExchangeName(name string) func(*ConsumeOptions)

WithConsumeOptionsBindingExchangeName returns a function that sets the exchange name the queue will be bound to

func WithConsumeOptionsBindingExchangeNoWait added in v0.6.0 

func WithConsumeOptionsBindingExchangeNoWait(options *ConsumeOptions)

WithConsumeOptionsBindingExchangeNoWait returns a function that sets the binding exchange noWait flag

func WithConsumeOptionsBindingNoWait added in v0.3.0 

func WithConsumeOptionsBindingNoWait(options *ConsumeOptions)

WithConsumeOptionsBindingNoWait sets the bindings to nowait, which means if the queue can not be bound the channel will not be closed with an error.

func WithConsumeOptionsConcurrency added in v0.3.0 

func WithConsumeOptionsConcurrency(concurrency int) func(*ConsumeOptions)

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

func WithConsumeOptionsConsumerExclusive added in v0.3.0 

func WithConsumeOptionsConsumerExclusive(options *ConsumeOptions)

WithConsumeOptionsConsumerExclusive 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 WithConsumeOptionsConsumerName added in v0.3.0 

func WithConsumeOptionsConsumerName(consumerName string) func(*ConsumeOptions)

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

func WithConsumeOptionsConsumerNoWait added in v0.3.0 

func WithConsumeOptionsConsumerNoWait(options *ConsumeOptions)

WithConsumeOptionsConsumerNoWait 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 WithConsumeOptionsQOSGlobal added in v0.3.0 

func WithConsumeOptionsQOSGlobal(options *ConsumeOptions)

WithConsumeOptionsQOSGlobal 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 WithConsumeOptionsQOSPrefetch added in v0.3.0 

func WithConsumeOptionsQOSPrefetch(prefetchCount int) func(*ConsumeOptions)

WithConsumeOptionsQOSPrefetch 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 WithConsumeOptionsQueueAutoDelete added in v0.3.0 

func WithConsumeOptionsQueueAutoDelete(options *ConsumeOptions)

WithConsumeOptionsQueueAutoDelete sets the queue to auto delete, which means it will be deleted when there are no more conusmers on it

func WithConsumeOptionsQueueDurable added in v0.3.0 

func WithConsumeOptionsQueueDurable(options *ConsumeOptions)

WithConsumeOptionsQueueDurable sets the queue to durable, which means it won't be destroyed when the server restarts. It must only be bound to durable exchanges

func WithConsumeOptionsQueueExclusive added in v0.3.0 

func WithConsumeOptionsQueueExclusive(options *ConsumeOptions)

WithConsumeOptionsQueueExclusive sets the queue to exclusive, which means it's are only accessible by the connection that declares it and will be deleted when the connection closes. Channels on other connections will receive an error when attempting to declare, bind, consume, purge or delete a queue with the same name.

func WithConsumeOptionsQueueNoWait added in v0.3.0 

func WithConsumeOptionsQueueNoWait(options *ConsumeOptions)

WithConsumeOptionsQueueNoWait sets the queue to nowait, which means the queue will assume to be declared on the server. A channel exception will arrive if the conditions are met for existing queues or attempting to modify an existing queue from a different connection.

func WithConsumeOptionsQuorum added in v0.3.0 

func WithConsumeOptionsQuorum(options *ConsumeOptions)

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

func WithConsumerOptionsLogger added in v0.5.0 

func WithConsumerOptionsLogger(log 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 sets a logger to log to stdout

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 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 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 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 consumer options

Types

type BindingExchangeOptions added in v0.6.0 

type BindingExchangeOptions struct {
	Name         string
	Kind         string
	Durable      bool
	AutoDelete   bool
	Internal     bool
	NoWait       bool
	ExchangeArgs Table
}

BindingExchangeOptions are used when binding to an exchange. it will verify the exchange is created before binding to it.

type ConsumeOptions 

type ConsumeOptions struct {
	QueueDurable      bool
	QueueAutoDelete   bool
	QueueExclusive    bool
	QueueNoWait       bool
	QueueArgs         Table
	BindingExchange   *BindingExchangeOptions
	BindingNoWait     bool
	BindingArgs       Table
	Concurrency       int
	QOSPrefetch       int
	QOSGlobal         bool
	ConsumerName      string
	ConsumerAutoAck   bool
	ConsumerExclusive bool
	ConsumerNoWait    bool
	ConsumerNoLocal   bool
	ConsumerArgs      Table
}

ConsumeOptions 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(url string, optionFuncs ...func(*ConsumerOptions)) (Consumer, error)

NewConsumer returns a new Consumer connected to the given rabbitmq server

func (Consumer) StartConsuming added in v0.1.0 

func (consumer Consumer) StartConsuming(
	handler func(d Delivery) bool,
	queue string,
	routingKeys []string,
	optionFuncs ...func(*ConsumeOptions),
) error

StartConsuming starts n goroutines where n="ConsumeOptions.QosOptions.Concurrency". Each goroutine spawns a handler that consumes off of the qiven queue which binds to the routing key(s). The provided handler is called once for each message. If the provided queue doesn't exist, it will be created on the cluster

func (Consumer) StopConsuming added in v0.6.0 

func (consumer Consumer) StopConsuming()

StopConsuming stops the consumption of messages. The consumer should be discarded as it's not safe for re-use

type ConsumerOptions 

type ConsumerOptions struct {
	Logging bool
	Logger  Logger
}

ConsumerOptions are used to describe a consumer's configuration. Logging set to true will enable the consumer to print to stdout Logger specifies a custom Logger interface implementation overruling Logging.

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 Logger added in v0.5.0 

type Logger interface {
	Printf(string, ...interface{})
}

Logger is the interface to send logs to. 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
	ContentType string
	// Transient or Persistent
	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
	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(url string, optionFuncs ...func(*PublisherOptions)) (Publisher, <-chan Return, 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) 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

type PublisherOptions added in v0.1.0 

type PublisherOptions struct {
	Logging bool
	Logger  Logger
}

PublisherOptions are used to describe a publisher's configuration. Logging set to true will enable the consumer to print to stdout

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 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
consumer
logger
publisher

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.