types

package
v0.21.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 25, 2026 License: Apache-2.0 Imports: 5 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	MQTTQueueName  = "mqtt"
	MQTTQueueTopic = "$queue/#"
)

Queue constants.

View Source 
const (
	// Queue delivery metadata properties.
	PropMessageID = "message-id"
	PropGroupID   = "group-id"
	PropQueueName = "queue"
	PropOffset    = "offset"

	// Stream delivery metadata properties.
	PropStreamOffset    = "x-stream-offset"
	PropStreamTimestamp = "x-stream-timestamp"

	// Work stealing metadata properties.
	PropWorkCommittedOffset = "x-work-committed-offset"
	PropWorkAcked           = "x-work-acked"
	PropWorkGroup           = "x-work-group"

	// Queue commit headers/properties.
	PropCommitGroupID = "x-group-id"
	PropCommitOffset  = "x-offset"

	// Queue reject metadata.
	PropRejectReason = "reason"

	// Internal queue forwarding metadata.
	PropForwardTargetQueues = "x-queue-forward-targets"
)

Variables

View Source 
var ErrInvalidConfig = errors.New("invalid queue configuration")

ErrInvalidConfig indicates an invalid queue configuration.

Functions

func ExtractQueueRoot 

func ExtractQueueRoot(topic string) string

ExtractQueueRoot extracts the queue root from a topic. Convention: $queue/{name} is the root, everything after is routing key. Example: "$queue/tasks/images" -> queue root is "$queue/tasks".

func ExtractRoutingKey 

func ExtractRoutingKey(topic, queueRoot string) string

RoutingKey extracts the routing key from a full topic. For topic "$queue/tasks/images/png", if queue root is "$queue/tasks", the routing key is "images/png".

func IsQueueWildcard 

func IsQueueWildcard(pattern string) bool

IsQueueWildcard returns true if the pattern contains wildcards.

func IsReservedQueueDeliveryProperty 

func IsReservedQueueDeliveryProperty(key string) bool

IsReservedQueueDeliveryProperty returns true for keys managed by queue routing.

Types

type Consumer 

type Consumer struct {
	ID            string
	ClientID      string
	GroupID       string
	QueueName     string
	RegisteredAt  time.Time
	LastHeartbeat time.Time
	ProxyNodeID   string // For cluster routing
}

Consumer represents a queue consumer.

type ConsumerGroup 

type ConsumerGroup struct {

	// Identity
	ID        string // Group identifier
	QueueName string // Queue this group consumes from
	Pattern   string // Subscription pattern (e.g., "sensors/#")
	Mode      ConsumerGroupMode

	// AutoCommit controls whether stream groups automatically commit offsets
	// as messages are delivered. Default is true for backwards compatibility.
	AutoCommit bool

	// Queue cursor state (single cursor per queue, no partitions)
	Cursor *QueueCursor

	// Pending Entry List (PEL) - messages delivered but not acked
	// Organized by consumer for efficient work stealing
	PEL map[string][]*PendingEntry // ConsumerID -> pending entries

	// Consumer membership
	Consumers map[string]*ConsumerInfo // ConsumerID -> consumer info

	// Timestamps
	CreatedAt time.Time
	UpdatedAt time.Time
	// contains filtered or unexported fields
}

ConsumerGroup represents the complete state of a consumer group. This includes cursor, PEL, and consumer membership. All map access is protected by an internal mutex for thread safety.

func NewConsumerGroupState 

func NewConsumerGroupState(queueName, groupID, pattern string) *ConsumerGroup

NewConsumerGroupState creates a new consumer group state.

func (*ConsumerGroup) AddPending 

func (g *ConsumerGroup) AddPending(consumerID string, entry *PendingEntry)

AddPending adds a pending entry for a consumer.

func (*ConsumerGroup) ConsumerCount 

func (g *ConsumerGroup) ConsumerCount() int

ConsumerCount returns the number of consumers in the group.

func (*ConsumerGroup) ConsumerIDs 

func (g *ConsumerGroup) ConsumerIDs() []string

ConsumerIDs returns a slice of all consumer IDs.

func (*ConsumerGroup) DeleteConsumer 

func (g *ConsumerGroup) DeleteConsumer(consumerID string)

DeleteConsumer removes a consumer by ID.

func (*ConsumerGroup) DeleteConsumerPEL 

func (g *ConsumerGroup) DeleteConsumerPEL(consumerID string)

DeleteConsumerPEL removes all pending entries for a consumer.

func (*ConsumerGroup) FindPending 

func (g *ConsumerGroup) FindPending(offset uint64) (*PendingEntry, string)

FindPending finds a pending entry by offset across all consumers.

func (*ConsumerGroup) ForEachConsumer 

func (g *ConsumerGroup) ForEachConsumer(fn func(id string, info *ConsumerInfo) bool)

ForEachConsumer iterates over all consumers with the lock held. Return false from fn to stop iteration.

func (*ConsumerGroup) GetConsumer 

func (g *ConsumerGroup) GetConsumer(consumerID string) *ConsumerInfo

GetConsumer returns a consumer by ID, or nil if not found.

func (*ConsumerGroup) GetCursor 

func (g *ConsumerGroup) GetCursor() *QueueCursor

GetCursor returns the queue cursor, creating if needed.

func (*ConsumerGroup) MinPendingOffset 

func (g *ConsumerGroup) MinPendingOffset() (uint64, bool)

MinPendingOffset returns the minimum offset across all PEL entries. This is used to calculate the committed offset.

func (*ConsumerGroup) PendingCount 

func (g *ConsumerGroup) PendingCount() int

PendingCount returns the total number of pending entries.

func (*ConsumerGroup) RemovePending 

func (g *ConsumerGroup) RemovePending(consumerID string, offset uint64) bool

RemovePending removes a pending entry for a consumer by offset.

func (*ConsumerGroup) ReplacePEL 

func (g *ConsumerGroup) ReplacePEL(pel map[string][]*PendingEntry)

ReplacePEL atomically replaces the entire PEL map.

func (*ConsumerGroup) SetConsumer 

func (g *ConsumerGroup) SetConsumer(consumerID string, info *ConsumerInfo)

SetConsumer adds or updates a consumer.

func (*ConsumerGroup) StealableEntries 

func (g *ConsumerGroup) StealableEntries(visibilityTimeout time.Duration, excludeConsumer string) []*PendingEntry

StealableEntries returns entries that are older than the visibility timeout.

func (*ConsumerGroup) TransferPending 

func (g *ConsumerGroup) TransferPending(offset uint64, fromConsumer, toConsumer string) bool

TransferPending moves a pending entry from one consumer to another.

type ConsumerGroupMode 

type ConsumerGroupMode string

ConsumerGroupMode defines how a consumer group is tracked. 

const (
	GroupModeQueue  ConsumerGroupMode = "queue"
	GroupModeStream ConsumerGroupMode = "stream"
)

type ConsumerInfo 

type ConsumerInfo struct {
	ID            string    // Consumer identifier (usually client ID)
	ClientID      string    // MQTT client ID
	ProxyNodeID   string    // Cluster node handling this consumer
	RegisteredAt  time.Time // When the consumer joined the group
	LastHeartbeat time.Time // Last activity timestamp
}

ConsumerInfo represents a consumer within a consumer group.

type CursorOption 

type CursorOption struct {
	Position CursorPosition
	Offset   uint64 // only used when Position == CursorOffset
	// Timestamp is used when Position == CursorTimestamp.
	Timestamp time.Time
	// Mode defines the consumer group mode (queue or stream).
	Mode ConsumerGroupMode
	// AutoCommit controls whether the consumer group automatically commits
	// offsets as messages are delivered. nil = default (true for streams),
	// explicit false = manual commit required.
	AutoCommit *bool
}

CursorOption specifies cursor positioning for SubscribeWithCursor.

type CursorPosition 

type CursorPosition int

CursorPosition defines the starting position for a consumer cursor. 

const (
	CursorDefault   CursorPosition = iota // resume from stored position
	CursorEarliest                        // start from beginning
	CursorLatest                          // start from end
	CursorOffset                          // start from specific offset
	CursorTimestamp                       // start from a timestamp
)

type DLQConfig 

type DLQConfig struct {
	Enabled      bool
	Topic        string
	AlertWebhook string
}

DLQConfig defines dead-letter queue configuration.

type DeliveryState 

type DeliveryState struct {
	MessageID   string
	QueueName   string
	GroupID     string // Consumer group ID - required for fan-out to multiple groups
	ConsumerID  string
	DeliveredAt time.Time
	Timeout     time.Time
	RetryCount  int
}

DeliveryState tracks inflight message delivery. Each consumer group has independent delivery tracking for fan-out support.

type Message 

type Message struct {
	ID         string
	Payload    []byte                 // Deprecated: Use PayloadBuf for zero-copy
	PayloadBuf *core.RefCountedBuffer // Zero-copy payload buffer (preferred)
	Topic      string
	Sequence   uint64
	Properties map[string]string

	// Lifecycle tracking
	State       MessageState
	CreatedAt   time.Time
	DeliveredAt time.Time
	NextRetryAt time.Time
	RetryCount  int

	// DLQ metadata
	FailureReason string
	FirstAttempt  time.Time
	LastAttempt   time.Time
	MovedToDLQAt  time.Time
	ExpiresAt     time.Time
}

Message represents a message in the queue system.

func (*Message) GetPayload 

func (m *Message) GetPayload() []byte

GetPayload returns the message payload, preferring PayloadBuf if available. This provides backward compatibility during migration to zero-copy.

func (*Message) IsExpired 

func (m *Message) IsExpired() bool

IsExpired reports whether the message has passed its expiry time. Messages with a zero ExpiresAt are considered non-expiring.

func (*Message) ReleasePayload 

func (m *Message) ReleasePayload()

ReleasePayload releases the buffer reference if PayloadBuf is set. This should be called when the message is no longer needed.

func (*Message) SetPayloadFromBuffer 

func (m *Message) SetPayloadFromBuffer(buf *core.RefCountedBuffer)

SetPayloadFromBuffer sets the payload from a RefCountedBuffer. The message takes ownership of one reference.

func (*Message) SetPayloadFromBytes 

func (m *Message) SetPayloadFromBytes(data []byte)

SetPayloadFromBytes creates a new buffer from bytes (for backward compatibility). This will eventually be phased out in favor of direct buffer creation.

type MessageState 

type MessageState string

MessageState represents the lifecycle state of a queue message. 

const (
	StateQueued    MessageState = "queued"
	StateDelivered MessageState = "delivered"
	StateAcked     MessageState = "acked"
	StateRetry     MessageState = "retry"
	StateDLQ       MessageState = "dlq"
)

type PendingEntry 

type PendingEntry struct {
	Offset        uint64    // Message offset in the queue log
	ConsumerID    string    // Consumer that claimed this entry
	ClaimedAt     time.Time // When the entry was claimed
	DeliveryCount int       // Number of times this message has been delivered
}

PendingEntry represents a message that has been delivered but not yet acknowledged. This is part of the PEL (Pending Entry List) for work stealing support.

type PublishMode 

type PublishMode int

PublishMode controls how the queue manager should handle a publish. 

const (
	PublishNormal PublishMode = iota
	PublishLocal
	PublishForwarded
)

type PublishRequest 

type PublishRequest struct {
	ClientID   string
	Topic      string
	Payload    []byte
	Properties map[string]string
}

PublishRequest encapsulates publish data for queue routing.

type QueueConfig 

type QueueConfig struct {
	Name     string
	Topics   []string // Topic patterns that route to this queue (e.g., "sensors/#", "orders/+/created")
	Reserved bool     // True for system queues like "mqtt" that cannot be deleted
	Type     QueueType

	// PrimaryGroup defines the consumer group whose committed offset is used
	// to report delivery status to stream consumers.
	PrimaryGroup string

	// Durability
	Durable                bool          // true = persists indefinitely, false = ephemeral (cleaned up when no consumers remain)
	ExpiresAfter           time.Duration // Grace period before ephemeral queue deletion (default 5m)
	LastConsumerDisconnect time.Time     // Zero value = has active consumers; set when last consumer leaves

	RetryPolicy RetryPolicy
	DLQConfig   DLQConfig
	Replication ReplicationConfig
	Retention   RetentionPolicy

	// Limits
	MaxMessageSize int64
	MaxDepth       int64
	MessageTTL     time.Duration

	// Performance
	DeliveryTimeout  time.Duration
	BatchSize        int
	HeartbeatTimeout time.Duration
}

QueueConfig defines configuration for a queue.

func DefaultEphemeralQueueConfig 

func DefaultEphemeralQueueConfig(name string, topics ...string) QueueConfig

DefaultEphemeralQueueConfig returns default ephemeral queue configuration.

func DefaultQueueConfig 

func DefaultQueueConfig(name string, topics ...string) QueueConfig

DefaultQueueConfig returns default queue configuration.

func FromInput 

func FromInput(input QueueConfigInput) QueueConfig

FromInput creates a QueueConfig from a simplified input config.

func MQTTQueueConfig 

func MQTTQueueConfig() QueueConfig

MQTTQueueConfig returns the reserved mqtt queue configuration.

func (*QueueConfig) Validate 

func (c *QueueConfig) Validate() error

Validate validates queue configuration.

type QueueConfigInput 

type QueueConfigInput struct {
	Name           string
	Topics         []string
	Reserved       bool
	Type           QueueType
	PrimaryGroup   string
	MaxMessageSize int64
	MaxDepth       int64
	MessageTTL     time.Duration
	MaxRetries     int
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	Multiplier     float64
	DLQEnabled     bool
	DLQTopic       string
	Retention      RetentionPolicy
	Replication    ReplicationConfig
}

QueueConfigInput is a simplified queue configuration from the main config file.

type QueueCursor 

type QueueCursor struct {
	Cursor    uint64 // Next offset to deliver (read position)
	Committed uint64 // Oldest unacknowledged offset (safe truncation point)
}

QueueCursor tracks consumption state for a queue within a consumer group.

type QueueType 

type QueueType string

QueueType defines the queue behavior mode. 

const (
	QueueTypeClassic QueueType = "classic"
	QueueTypeStream  QueueType = "stream"
)

type ReplicationConfig 

type ReplicationConfig struct {
	Enabled           bool
	Group             string          // Logical Raft group ID for this queue (empty = default)
	ReplicationFactor int             // Number of replicas (default: 3)
	Mode              ReplicationMode // sync or async
	MinInSyncReplicas int             // Min replicas that must ACK (default: 2)
	AckTimeout        time.Duration   // Timeout for sync mode operations (default: 5s)

	// Raft tuning (optional, uses defaults if zero)
	HeartbeatTimeout  time.Duration // Raft heartbeat interval (default: 1s)
	ElectionTimeout   time.Duration // Raft election timeout (default: 3s)
	SnapshotInterval  time.Duration // Snapshot frequency (default: 5m)
	SnapshotThreshold uint64        // Snapshot after N log entries (default: 8192)
}

ReplicationConfig defines Raft-based replication for queues.

type ReplicationMode 

type ReplicationMode string

ReplicationMode defines the replication behavior for queue messages. 

const (
	ReplicationSync  ReplicationMode = "sync"  // Wait for quorum ACK before returning
	ReplicationAsync ReplicationMode = "async" // Return immediately after leader accepts
)

type RetentionPolicy 

type RetentionPolicy struct {
	// Time-based retention (background cleanup)
	RetentionTime     time.Duration // Delete messages older than this (0 = disabled)
	TimeCheckInterval time.Duration // How often to run cleanup (default: 5m)

	// Size-based retention (active check on enqueue)
	RetentionBytes    int64 // Max total queue size in bytes (0 = unlimited)
	RetentionMessages int64 // Max message count (0 = unlimited)
	SizeCheckEvery    int   // Check size every N enqueues (default: 100, optimization)

	// Log compaction (Kafka-style)
	CompactionEnabled  bool          // Enable log compaction
	CompactionKey      string        // Message property to use as compaction key
	CompactionLag      time.Duration // Wait before compacting new messages (default: 5m)
	CompactionInterval time.Duration // How often to run compaction (default: 10m)
}

RetentionPolicy defines Kafka-style retention policies for automatic message cleanup.

type RetryPolicy 

type RetryPolicy struct {
	MaxRetries        int
	InitialBackoff    time.Duration
	MaxBackoff        time.Duration
	BackoffMultiplier float64
	TotalTimeout      time.Duration
}

RetryPolicy defines retry behavior for failed messages.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.