tree

Documentation

Index

• type QueueTree
  • func New(logger log.Logger, scope metrics.Scope, partitions []string, ...) (*QueueTree, error)
  • func (t *QueueTree) Enqueue(ctx context.Context, items []types.Item) ([]types.ItemToPersist, error)
  • func (t *QueueTree) Start(ctx context.Context) error
  • func (t *QueueTree) Stop(ctx context.Context) error
  • func (t *QueueTree) String() string
• type QueueTreeNode
  • func (n *QueueTreeNode) Enqueue(ctx context.Context, item types.Item, partitions []string, ...) (types.ItemToPersist, error)
  • func (n *QueueTreeNode) Init(logger log.Logger, scope metrics.Scope, policyCol types.NodePolicyCollection, ...) error
  • func (n *QueueTreeNode) Start(ctx context.Context, consumerFactory types.ConsumerFactory, ...) error
  • func (n *QueueTreeNode) Stop(ctx context.Context) error
  • func (n *QueueTreeNode) String() string

Types

type QueueTree

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

QueueTree is a tree structure that represents the queue structure for MAPQ

func New

func New(
	logger log.Logger,
	scope metrics.Scope,
	partitions []string,
	policies []types.NodePolicy,
	persister types.Persister,
	consumerFactory types.ConsumerFactory,
) (*QueueTree, error)

func (*QueueTree) Enqueue

func (t *QueueTree) Enqueue(ctx context.Context, items []types.Item) ([]types.ItemToPersist, error)

func (*QueueTree) Start

func (t *QueueTree) Start(ctx context.Context) error

Start the dispatchers for all leaf nodes

func (*QueueTree) Stop

func (t *QueueTree) Stop(ctx context.Context) error

Stop the dispatchers for all leaf nodes

func (*QueueTree) String

func (t *QueueTree) String() string

type QueueTreeNode

type QueueTreeNode struct {

	// The path to the node
	Path string

	// The partition key used by this node to decide which child to enqueue the item.
	// Partition key of a node is the attribute key of child node.
	PartitionKey string

	// The attribute key used to create this node by parent
	AttributeKey string

	// The attribute value used to create this node by parent
	AttributeVal any

	// The policy for this node. It's merged policy from all policies that match this node
	NodePolicy types.NodePolicy

	// Children by attribute key
	// "*" is a special key that represents the default/fallback child queue
	// If there's no children then the node is considered leaf node
	Children map[any]*QueueTreeNode

	// The dispatcher for this node. Only leaf nodes have dispatcher
	Dispatcher *dispatcher.Dispatcher
	// contains filtered or unexported fields
}

QueueTreeNode represents a node in the queue tree

func (*QueueTreeNode) Enqueue

func (n *QueueTreeNode) Enqueue(
	ctx context.Context,
	item types.Item,
	partitions []string,
	partitionMap map[string]any,
) (types.ItemToPersist, error)

func (*QueueTreeNode) Init

func (n *QueueTreeNode) Init(logger log.Logger, scope metrics.Scope, policyCol types.NodePolicyCollection, partitions []string) error

func (*QueueTreeNode) Start

func (n *QueueTreeNode) Start(
	ctx context.Context,
	consumerFactory types.ConsumerFactory,
	partitions []string,
	partitionMap map[string]any,
) error

func (*QueueTreeNode) Stop

func (n *QueueTreeNode) Stop(ctx context.Context) error

func (*QueueTreeNode) String

func (n *QueueTreeNode) String() string