queue

Documentation

Index

• type Queue
  • func NewQueue(name string, connID uint64, exclusive bool, autoDelete bool, durable bool, ...) *Queue
  • func (queue *Queue) AckMsg(message *amqp.Message)
  • func (queue *Queue) AddConsumer(consumer interfaces.Consumer, exclusive bool) error
  • func (queue *Queue) ConnID() uint64
  • func (queue *Queue) ConsumersCount() int
  • func (queue *Queue) Delete(ifUnused bool, ifEmpty bool) (uint64, error)
  • func (queue *Queue) EqualWithErr(qB *Queue) error
  • func (queue *Queue) GetName() string
  • func (queue *Queue) IsActive() bool
  • func (queue *Queue) IsAutoDelete() bool
  • func (queue *Queue) IsDurable() bool
  • func (queue *Queue) IsExclusive() bool
  • func (queue *Queue) Length() uint64
  • func (queue *Queue) Marshal(protoVersion string) (data []byte, err error)
  • func (queue *Queue) Pop() *amqp.Message
  • func (queue *Queue) PopQos(qosList []*qos.AmqpQos) *amqp.Message
  • func (queue *Queue) Purge() (length uint64)
  • func (queue *Queue) Push(message *amqp.Message)
  • func (queue *Queue) RemoveConsumer(cTag string)
  • func (queue *Queue) Requeue(message *amqp.Message)
  • func (queue *Queue) Start() error
  • func (queue *Queue) Stop() error
  • func (queue *Queue) Unmarshal(data []byte, protoVersion string) (err error)

Types

type Queue

type Queue struct {
	safequeue.SafeQueue
	// contains filtered or unexported fields
}

Queue is an implementation of the AMQP-queue entity

func NewQueue

func NewQueue(name string, connID uint64, exclusive bool, autoDelete bool, durable bool, config config.Queue, msgStorageP interfaces.MsgStorage, msgStorageT interfaces.MsgStorage, autoDeleteQueue chan string) *Queue

NewQueue returns new instance of Queue

func (*Queue) AckMsg

func (queue *Queue) AckMsg(message *amqp.Message)

AckMsg accept ack event for message

func (*Queue) AddConsumer

func (queue *Queue) AddConsumer(consumer interfaces.Consumer, exclusive bool) error

AddConsumer add consumer to consumer messages with exclusive check

func (*Queue) ConnID

func (queue *Queue) ConnID() uint64

ConnID returns ID of connection that create this queue

func (*Queue) ConsumersCount

func (queue *Queue) ConsumersCount() int

ConsumersCount returns consumers count

func (*Queue) Delete

func (queue *Queue) Delete(ifUnused bool, ifEmpty bool) (uint64, error)

Delete cancel consumers and delete its messages from storage

func (*Queue) EqualWithErr

func (queue *Queue) EqualWithErr(qB *Queue) error

EqualWithErr returns is given queue equal to current

func (*Queue) GetName

func (queue *Queue) GetName() string

GetName returns queue name

func (*Queue) IsActive

func (queue *Queue) IsActive() bool

IsActive returns is queue's main loop is active

func (*Queue) IsAutoDelete

func (queue *Queue) IsAutoDelete() bool

IsAutoDelete returns is queue should be deleted automatically

func (*Queue) IsDurable

func (queue *Queue) IsDurable() bool

IsDurable returns is queue durable

func (*Queue) IsExclusive

func (queue *Queue) IsExclusive() bool

IsExclusive returns is queue exclusive

func (*Queue) Length

func (queue *Queue) Length() uint64

Length returns queue length

func (*Queue) Marshal

func (queue *Queue) Marshal(protoVersion string) (data []byte, err error)

Marshal returns raw representation of queue to store into storage

func (*Queue) Pop

func (queue *Queue) Pop() *amqp.Message

Pop returns message from queue head without QOS check

func (*Queue) PopQos

func (queue *Queue) PopQos(qosList []*qos.AmqpQos) *amqp.Message

PopQos returns message from queue head with QOS check

func (*Queue) Purge

func (queue *Queue) Purge() (length uint64)

Purge clean queue and message storage for durable queues

func (*Queue) Push

func (queue *Queue) Push(message *amqp.Message)

Push append message into queue tail and put it into message storage if queue is durable and message's persistent flag is true

func (*Queue) RemoveConsumer

func (queue *Queue) RemoveConsumer(cTag string)

RemoveConsumer remove consumer If it was last consumer and queue is auto-delete - queue will be removed

func (*Queue) Requeue

func (queue *Queue) Requeue(message *amqp.Message)

Requeue add message into queue head

func (*Queue) Start

func (queue *Queue) Start() error

Start starts base queue loop to send events to consumers Current consumer to handle message from queue selected by round robin

func (*Queue) Stop

func (queue *Queue) Stop() error

Stop stops main queue loop After stop no one can send or receive messages from queue

func (*Queue) Unmarshal

func (queue *Queue) Unmarshal(data []byte, protoVersion string) (err error)

Unmarshal returns queue from storage raw bytes data