queue

Documentation

Index

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

Constants

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

const DefaultNumWorkers = 50

const HighPriority = Priority(10)

const LowPriority = Priority(1)

const MediumPriority = Priority(5)

Functions

func CreateQueueWorkers

func CreateQueueWorkers(ctx context.Context, numWorks uint64, queue network.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 struct {
	// contains filtered or unexported fields
}

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

func NewMessageQueue

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

func (*MessageQueue) Insert

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

func (*MessageQueue) Len

func (mq *MessageQueue) Len() int

func (*MessageQueue) Remove

func (mq *MessageQueue) 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 QMessage

type QMessage struct {
	Payload  interface{}      // the decoded message
	Size     int              // the size of the message in bytes
	Target   channels.Channel // the target channel to lookup the engine
	SenderID flow.Identifier  // senderID for logging
}

QMessage is the message that is enqueued for each incoming message