queue

Documentation

Index

• Constants
• func CreateQueueWorkers(ctx context.Context, numWorks uint64, queue MessageQueue, ...)
• type MessagePriorityFunc
• type MessageQueue
• type MessageQueueImpl
  • func NewMessageQueue(ctx context.Context, priorityFunc MessagePriorityFunc, ...) *MessageQueueImpl
  • func (mq *MessageQueueImpl) Insert(message interface{}) error
  • func (mq *MessageQueueImpl) Len() int
  • func (mq *MessageQueueImpl) Remove() interface{}
• type Priority
  • func GetEventPriority(message interface{}) (Priority, error)
• type QueueMessage

Constants

const (
	KiB = 1 << (10 * iota)
	MiB
)

const DefaultNumWorkers = 5

const HighPriority = Priority(10)

const LowPriority = Priority(1)

const MediumPriority = Priority(5)

Functions

func CreateQueueWorkers

func CreateQueueWorkers(ctx context.Context, numWorks uint64, queue MessageQueue, callback func(interface{}))

CreateQueueWorkers creates queue workers to read from the queue

Types

type MessagePriorityFunc

type MessagePriorityFunc func(message interface{}) (Priority, error)

MessagePriorityFunc - the callback function to derive priority of a message

type MessageQueue

type MessageQueue interface {
	// Insert inserts the message in queue
	Insert(message interface{}) error
	// Remove removes the message from the queue in priority order. If no message is found, this call blocks.
	// If two messages have the same priority, items are de-queued in insertion order
	Remove() interface{}
	// Len gives the current length of the queue
	Len() int
}

MessageQueue is the interface of the inbound message queue

type MessageQueueImpl

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

MessageQueueImpl is the heap based priority queue implementation of the MessageQueue implementation

func NewMessageQueue

func NewMessageQueue(ctx context.Context, priorityFunc MessagePriorityFunc, nm module.NetworkMetrics) *MessageQueueImpl

func (*MessageQueueImpl) Insert

func (mq *MessageQueueImpl) Insert(message interface{}) error

func (*MessageQueueImpl) Len

func (mq *MessageQueueImpl) Len() int

func (*MessageQueueImpl) Remove

func (mq *MessageQueueImpl) Remove() interface{}

type Priority

type Priority int

func GetEventPriority

func GetEventPriority(message interface{}) (Priority, error)

GetEventPriority returns the priority of the flow event message. It is an average of the priority by message type and priority by message size

type QueueMessage

type QueueMessage struct {
	Payload   interface{}     // the decoded message
	Size      int             // the size of the message in bytes
	ChannelID string          // the channel id to use to lookup the engine
	SenderID  flow.Identifier // senderID for logging
}

QueueMessage is the message that is enqueued for each incoming message