consumer

Documentation

Index

• type Consumer
  • func NewConsumer(config *models.RabbitSeasoning, channelPool *pools.ChannelPool, ...) (*Consumer, error)
  • func NewConsumerFromConfig(config *models.ConsumerConfig, channelPool *pools.ChannelPool) (*Consumer, error)
  • func (con *Consumer) Errors() <-chan error
  • func (con *Consumer) FlushErrors()
  • func (con *Consumer) FlushMessages()
  • func (con *Consumer) FlushStop()
  • func (con *Consumer) Get(queueName string, autoAck bool) (*models.Message, error)
  • func (con *Consumer) GetBatch(queueName string, batchSize int, autoAck bool) ([]*models.Message, error)
  • func (con *Consumer) Messages() <-chan *models.Message
  • func (con *Consumer) StartConsuming() error
  • func (con *Consumer) StopConsuming(immediate bool, flushMessages bool) error

Types

type Consumer

type Consumer struct {
	Config *models.RabbitSeasoning

	Enabled      bool
	QueueName    string
	ConsumerName string
	// contains filtered or unexported fields
}

Consumer receives messages from a RabbitMQ location.

func NewConsumer

func NewConsumer(
	config *models.RabbitSeasoning,
	channelPool *pools.ChannelPool,
	queuename string,
	consumerName string,
	autoAck bool,
	exclusive bool,
	noWait bool,
	args map[string]interface{},
	qosCountOverride int,
	messageBuffer uint32,
	errorBuffer uint32,
	sleepOnErrorInterval uint32,
	sleepOnIdleInterval uint32) (*Consumer, error)

NewConsumer creates a new Consumer to receive messages from a specific queuename.

func NewConsumerFromConfig

func NewConsumerFromConfig(
	config *models.ConsumerConfig,
	channelPool *pools.ChannelPool) (*Consumer, error)

NewConsumerFromConfig creates a new Consumer to receive messages from a specific queuename.

func (*Consumer) Errors

func (con *Consumer) Errors() <-chan error

Errors yields all the internal errs for consuming messages.

func (*Consumer) FlushErrors

func (con *Consumer) FlushErrors()

FlushErrors allows you to flush out all previous Errors.

func (*Consumer) FlushMessages

func (con *Consumer) FlushMessages()

FlushMessages allows you to flush out all previous Messages. WARNING: THIS WILL RESULT IN LOST MESSAGES.

func (*Consumer) FlushStop

func (con *Consumer) FlushStop()

FlushStop allows you to flush out all previous Stop signals.

func (*Consumer) Get

func (con *Consumer) Get(queueName string, autoAck bool) (*models.Message, error)

Get gets a single message from any queue.

func (*Consumer) GetBatch

func (con *Consumer) GetBatch(queueName string, batchSize int, autoAck bool) ([]*models.Message, error)

GetBatch gets a group of messages from any queue.

func (*Consumer) Messages

func (con *Consumer) Messages() <-chan *models.Message

Messages yields all the internal messages ready for consuming.

func (*Consumer) StartConsuming

func (con *Consumer) StartConsuming() error

StartConsuming starts the Consumer.

func (*Consumer) StopConsuming

func (con *Consumer) StopConsuming(immediate bool, flushMessages bool) error

StopConsuming allows you to signal stop to the consumer. Will stop on the consumer channelclose or responding to signal after getting all remaining deviveries. FlushMessages empties the internal buffer of messages received by queue. Ackable messages are still in RabbitMQ queue, while noAck messages will unfortunately be lost. Use wisely.