amqputil

Documentation

Overview

Package amqputil provides AmqpContext to simplify AMQP interaction

Index

• Variables
• type AmqpAccessor
• type AmqpConnectionHelper
  • func (helper *AmqpConnectionHelper) GetAmqpContext(consumerId string) (amqpContext *AmqpContext)
• type AmqpContext
  • func (amqpContext *AmqpContext) Channel() ChannelAccessor
  • func (amqpContext *AmqpContext) Close() error
  • func (amqpContext *AmqpContext) EnsureQueueExists(queueName string) error
  • func (amqpContext *AmqpContext) LastError() error
  • func (amqpContext *AmqpContext) PublishMessage(queueName string, message interface{}) error
  • func (amqpContext *AmqpContext) ReceiveMessage(queueName string, message interface{}) (delivery *amqp.Delivery, err error)
  • func (amqpContext *AmqpContext) ReceiveProtoMessage(queueName string, message proto.Message) (delivery *amqp.Delivery, err error)
  • func (amqpContext *AmqpContext) Reset() error
  • func (amqpContext *AmqpContext) ResetError()
  • func (amqpContext *AmqpContext) SetLastError(err error)
• type ChannelAccessor

Variables

var ErrNoMessage = errors.Errorf("No message found in queue")

ErrNoMessages indicates, that no message were found in a queue

Types

type AmqpAccessor

type AmqpAccessor interface {
	PublishMessage(queueName string, message interface{}) error
	ReceiveMessage(queueName string, message interface{}) (delivery *amqp.Delivery, err error)
	Channel() ChannelAccessor
	Close() error
	Reset() error
	LastError() error
	SetLastError(err error)
	ResetError()
}

type AmqpConnectionHelper

type AmqpConnectionHelper struct {
	AmqpConnectionURL string
}

AmqpConnectionHelper helps to get a connection AMQP

func (*AmqpConnectionHelper) GetAmqpContext

func (helper *AmqpConnectionHelper) GetAmqpContext(consumerId string) (amqpContext *AmqpContext)

GetAmqpContext creates an AmqpContext for the given amqpConnectionURL or returns an already existing AmqpContext for the amqpConnectionURL the consumerId identifies the consumer on the channel

type AmqpContext

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

AmqpContext simplifies amqp interaction by providing a context with a persistent connection and a channel to simplify message publishing

func (*AmqpContext) Channel

func (amqpContext *AmqpContext) Channel() ChannelAccessor

func (*AmqpContext) Close

func (amqpContext *AmqpContext) Close() error

Close closes the amqp connection

func (*AmqpContext) EnsureQueueExists

func (amqpContext *AmqpContext) EnsureQueueExists(queueName string) error

func (*AmqpContext) LastError

func (amqpContext *AmqpContext) LastError() error

func (*AmqpContext) PublishMessage

func (amqpContext *AmqpContext) PublishMessage(queueName string, message interface{}) error

PublishMessage sends given message as application/json to queue with given name. If the queue does not exist, it is created. Errors go to AmqpContext.Err

func (*AmqpContext) ReceiveMessage

func (amqpContext *AmqpContext) ReceiveMessage(queueName string, message interface{}) (delivery *amqp.Delivery, err error)

ReceiveMessage gets next message from queue with given queue name

func (*AmqpContext) ReceiveProtoMessage

func (amqpContext *AmqpContext) ReceiveProtoMessage(queueName string, message proto.Message) (delivery *amqp.Delivery, err error)

ReceiveMessage gets next message from queue with given queue name

func (*AmqpContext) Reset

func (amqpContext *AmqpContext) Reset() error

Reset resets the channel and queues - asumes that

func (*AmqpContext) ResetError

func (amqpContext *AmqpContext) ResetError()

func (*AmqpContext) SetLastError

func (amqpContext *AmqpContext) SetLastError(err error)

type ChannelAccessor

type ChannelAccessor interface {
	Qos(prefetchCount, prefetchSize int, global bool) error
	QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)
	Publish(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error
	Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)
	Close() error
	Cancel(consumer string, noWait bool) error
	QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, error)
	QueueInspect(name string) (amqp.Queue, error)
}

ChannelAccessor is an interface for the necessary methods to access the Channel struct of the AMQP library. the library does not define an interface, so we do it here (it helps for mocking) this interface only defines those methods that we know we need. See https://godoc.org/github.com/streadway/amqp for all possible methods.