rabbitClient

README

Rabbit Client

A "topic" only rabbit-client with thoughtful defaults and hopefully good abstraction (We try to do better)

Features

• Publisher that uses routing_key and payload
• Single subscriber can have multiple subscriptions
• Blocking and non-blocking subscriptions

How to use this

Start by looking at examples

Documentation

Index

• Constants
• type Publisher
  • func DialAsPublisher(connectionString, exchangeName string) *Publisher
  • func (publisher *Publisher) Close()
  • func (publisher *Publisher) PublishMessage(routingKey, message string)
• type Subscriber
  • func DialAsSubscriber(connectionString, exchangeName string) *Subscriber
  • func (subscriber *Subscriber) Close()
  • func (subscriber *Subscriber) EnsureSubscription(name, bindingKey string, exclusive bool, prefetch int) *Subscription
• type Subscription
  • func (subscription *Subscription) BlockExecution()
  • func (subscription *Subscription) BlockingListen(listener func([]byte))
  • func (subscription *Subscription) Listen(listener func([]byte))

Constants

const (
	// ExchangeIsDurable defaults to true to ensure broker survives restart. Otherwise, a re-declaraction is required.
	ExchangeIsDurable = true

	// ExchangeType defaults to "topic", because topic is versatile to be either fanout or direct based on the binding and routing keypair
	ExchangeType = "topic"

	// QueueIsDurable defaults to true to ensure persistent messages within the queue are not lost if queue is restarted
	QueueIsDurable = true

	// PublishIsMandatory defaults to true to ensure message is actually delivered.
	// Message could be dropped if mandatory=false and there is no queue with the binding with the routingKey
	PublishIsMandatory = true

	// DeliveryMode defaults to 2, which means message is persistent and always get persisted to harddisk
	DeliveryMode = amqp.Persistent
)

Types

type Publisher

type Publisher struct {
	Connection      *amqp.Connection
	OutgoingChannel *amqp.Channel
	ConnectionInfo  map[string]string
}

func DialAsPublisher

func DialAsPublisher(connectionString, exchangeName string) *Publisher

DialAsPublisher establishes a connection to broker and declare the exchange to use for the returned `Publisher` AMQP is too flexible. we reduce that flexibility and predictability by hard-fixing exchange name throughout an application lifecycle We should not see a single script handling multiple exchanges. If it does, it should be a separate TCP connection

func (*Publisher) Close

func (publisher *Publisher) Close()

func (*Publisher) PublishMessage

func (publisher *Publisher) PublishMessage(routingKey, message string)

type Subscriber

type Subscriber struct {
	Connection     *amqp.Connection
	ConnectionInfo map[string]string
}

func DialAsSubscriber

func DialAsSubscriber(connectionString, exchangeName string) *Subscriber

DialAsSubscriber establishes a connection to broker and declare the exchange to use for the returned `Subscriber` AMQP is too flexible. we reduce that flexibility and predictability by hard-fixing exchange name throughout an application lifecycle We should not see a single script handling multiple exchanges. If it does, it should be a separate TCP connection

func (*Subscriber) Close

func (subscriber *Subscriber) Close()

func (*Subscriber) EnsureSubscription

func (subscriber *Subscriber) EnsureSubscription(name, bindingKey string, exclusive bool, prefetch int) *Subscription

type Subscription

type Subscription struct {
	Subscriber *Subscriber
	Channel    *amqp.Channel
	Queue      *amqp.Queue
	BindingKey string
	Exclusive  bool
}

func (*Subscription) BlockExecution

func (subscription *Subscription) BlockExecution()

BlockExecution blocks the calling goroutine until os.Interrupt is fired

func (*Subscription) BlockingListen

func (subscription *Subscription) BlockingListen(listener func([]byte))

BlockingListen blocks the goroutine from continuing further

func (*Subscription) Listen

func (subscription *Subscription) Listen(listener func([]byte))

Listen without blocking the calling goroutine