consumer

Documentation

Index

• type Consumer
  • func NewConsumer(msgStore *msgstore.MessageStore, arguments *amqp.Table, ...) *Consumer
  • func (consumer *Consumer) AcquireResources(qm *amqp.QueueMessage) bool
  • func (consumer *Consumer) ConsumeImmediate(qm *amqp.QueueMessage, msg *amqp.Message) bool
  • func (consumer *Consumer) MarshalJSON() ([]byte, error)
  • func (consumer *Consumer) MessageResourceHolders() []amqp.MessageResourceHolder
  • func (consumer *Consumer) Ping()
  • func (consumer *Consumer) Redeliver(tag uint64, qm *amqp.QueueMessage)
  • func (consumer *Consumer) ReleaseResources(qm *amqp.QueueMessage)
  • func (consumer *Consumer) SendCancel()
  • func (consumer *Consumer) Start()
  • func (consumer *Consumer) Stop()
• type ConsumerChannel
• type ConsumerQueue

Types

type Consumer

type Consumer struct {
	ConsumerTag string

	StatCount uint64
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(
	msgStore *msgstore.MessageStore,
	arguments *amqp.Table,
	cchannel ConsumerChannel,
	consumerTag string,
	exclusive bool,
	noAck bool,
	noLocal bool,
	cqueue ConsumerQueue,
	queueName string,
	prefetchSize uint32,
	prefetchCount uint16,
	localId int64,
) *Consumer

func (*Consumer) AcquireResources

func (consumer *Consumer) AcquireResources(qm *amqp.QueueMessage) bool

func (*Consumer) ConsumeImmediate

func (consumer *Consumer) ConsumeImmediate(qm *amqp.QueueMessage, msg *amqp.Message) bool

func (*Consumer) MarshalJSON

func (consumer *Consumer) MarshalJSON() ([]byte, error)

func (*Consumer) MessageResourceHolders

func (consumer *Consumer) MessageResourceHolders() []amqp.MessageResourceHolder

TODO: make this a field that we construct on init

func (*Consumer) Ping

func (consumer *Consumer) Ping()

func (*Consumer) Redeliver

func (consumer *Consumer) Redeliver(tag uint64, qm *amqp.QueueMessage)

Send again, leave all stats the same since this consumer was already dealing with this message

func (*Consumer) ReleaseResources

func (consumer *Consumer) ReleaseResources(qm *amqp.QueueMessage)

func (*Consumer) SendCancel

func (consumer *Consumer) SendCancel()

func (*Consumer) Start

func (consumer *Consumer) Start()

func (*Consumer) Stop

func (consumer *Consumer) Stop()

type ConsumerChannel

type ConsumerChannel interface {
	amqp.MessageResourceHolder
	SendContent(method amqp.MethodFrame, msg *amqp.Message)
	SendMethod(method amqp.MethodFrame)
	FlowActive() bool
	AddUnackedMessage(consumerTag string, qm *amqp.QueueMessage, queueName string) uint64
}

The methods necessary for a consumer to interact with a channel

type ConsumerQueue

type ConsumerQueue interface {
	GetOne(rhs ...amqp.MessageResourceHolder) (*amqp.QueueMessage, *amqp.Message)
	MaybeReady() chan bool
}