Documentation
¶
Overview ¶
Package amqputil provides AmqpContext to simplify AMQP interaction
Index ¶
- Variables
- type AmqpAccessor
- type AmqpConnectionHelper
- 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
Constants ¶
This section is empty.
Variables ¶
var ErrNoMessage = errors.Errorf("No message found in queue")
ErrNoMessages indicates, that no message were found in a queue
Functions ¶
This section is empty.
Types ¶
type AmqpAccessor ¶
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 ¶ added in v0.0.9
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 ¶ added in v0.0.10
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://pkg.go.dev/github.com/rabbitmq/amqp091-go for all possible methods.
Source Files