streamqueue

Documentation

Overview

Package streamqueue defines package-owned stream queue semantics shared by native backend adapters.

Index

• Constants
• Variables
• func ApplyMessageMetadata(record *management.JobRecord, metadata *job.Metadata)
• func MessageAge(id string, now time.Time) (time.Duration, error)
• func MessageMetadata(body []byte) *job.Metadata
• type AckRequest
  • func (r AckRequest) Validate() error
• type AddRequest
  • func (r AddRequest) Validate(maxPayloadBytes int) error
• type ClaimRequest
  • func (r ClaimRequest) Validate() error
• type ClaimResult
• type DeadLetterRequest
  • func (r DeadLetterRequest) Validate(maxPayloadBytes int) error
• type Delivery
• type FailureMetadata
  • func (m FailureMetadata) Validate() error
• type GroupState
  • func (s GroupState) Stats() Stats
• type ReadRequest
  • func (r ReadRequest) Validate() error
• type RequestError
  • func (e *RequestError) Error() string
  • func (e *RequestError) Unwrap() []error
• type Stats
• type Transport

Constants

const MaxBatchSize int64 = 256

MaxBatchSize bounds one read or reclaim operation.

Variables

var (
	// ErrInvalidSemanticRequest classifies invalid stream command requests.
	ErrInvalidSemanticRequest = errors.New("streamqueue: invalid semantic request")
	// ErrMalformedDelivery classifies invalid backend delivery metadata.
	ErrMalformedDelivery = errors.New("streamqueue: malformed delivery")
)

Functions

func ApplyMessageMetadata

func ApplyMessageMetadata(record *management.JobRecord, metadata *job.Metadata)

ApplyMessageMetadata copies only the public allowlist into a management record. Backend-derived source identity remains separate from OriginalID.

func MessageAge

func MessageAge(id string, now time.Time) (time.Duration, error)

MessageAge derives an entry age from a server-generated stream identifier.

func MessageMetadata

func MessageMetadata(body []byte) *job.Metadata

MessageMetadata returns validated operational metadata from the bounded job envelope. Malformed and legacy bodies have no trustworthy metadata.

Types

type AckRequest

type AckRequest struct {
	Stream string
	Group  string
	ID     string
}

AckRequest identifies one consumer-group delivery to settle.

func (AckRequest) Validate

func (r AckRequest) Validate() error

Validate checks acknowledgement identity.

type AddRequest

type AddRequest struct {
	Stream    string
	MaxLength int64
	Body      []byte
}

AddRequest describes a bounded stream append.

func (AddRequest) Validate

func (r AddRequest) Validate(maxPayloadBytes int) error

Validate checks append ownership and resource bounds.

type ClaimRequest

type ClaimRequest struct {
	Stream   string
	Group    string
	Consumer string
	MinIdle  time.Duration
	Start    string
	Count    int64
}

ClaimRequest describes one bounded stale-delivery reclaim scan.

func (ClaimRequest) Validate

func (r ClaimRequest) Validate() error

Validate checks reclaim ownership and scan bounds.

type ClaimResult

type ClaimResult struct {
	Next       string
	Deliveries []Delivery
}

ClaimResult contains reclaimed entries and the next bounded scan cursor.

type DeadLetterRequest

type DeadLetterRequest struct {
	Source      string
	Destination string
	Group       string
	Delivery    Delivery
	Failure     FailureMetadata
}

DeadLetterRequest describes an append-before-ack terminal transfer.

func (DeadLetterRequest) Validate

func (r DeadLetterRequest) Validate(maxPayloadBytes int) error

Validate checks terminal transfer identity and payload bounds.

type Delivery

type Delivery struct {
	ID                   string
	Body                 []byte
	Attempts             int64
	Reclaimed            bool
	OriginalDeadLetterID string
	PriorDeadLetterID    string
	ReplayGeneration     uint32
}

Delivery is the transport-neutral representation of one stream entry.

type FailureMetadata

type FailureMetadata struct {
	Classification management.Classification
	Code           string
}

FailureMetadata is the bounded disposition persisted with a failure record.

func (FailureMetadata) Validate

func (m FailureMetadata) Validate() error

Validate rejects unknown classifications and unsafe codes.

type GroupState

type GroupState struct {
	Pending         int64
	Lag             int64
	OldestPendingID string
}

GroupState contains server-reported consumer-group depth components.

func (GroupState) Stats

func (s GroupState) Stats() Stats

Stats derives depth only when the server reports group lag.

type ReadRequest

type ReadRequest struct {
	Stream   string
	Group    string
	Consumer string
	Count    int64
	Block    time.Duration
}

ReadRequest describes one cancellation-aware consumer-group read.

func (ReadRequest) Validate

func (r ReadRequest) Validate() error

Validate checks consumer identity and command bounds.

type RequestError

type RequestError struct {
	Command string
	Field   string
	Cause   error
}

RequestError identifies an invalid semantic field without including its potentially sensitive value.

func (*RequestError) Error

func (e *RequestError) Error() string

Error returns value-free request validation text.

func (*RequestError) Unwrap

func (e *RequestError) Unwrap() []error

Unwrap retains the stable classification and underlying cause.

type Stats

type Stats struct {
	Depth    int64
	Pending  int64
	Lag      int64
	LagKnown bool
}

Stats is the common honest consumer-group depth snapshot.

type Transport

type Transport interface {
	EnsureGroup(context.Context, string, string) error
	Add(context.Context, AddRequest) (string, error)
	Read(context.Context, ReadRequest) ([]Delivery, error)
	Claim(context.Context, ClaimRequest) (ClaimResult, error)
	Ack(context.Context, AckRequest) error
	DeadLetter(context.Context, DeadLetterRequest) error
	GroupState(context.Context, string, string) (GroupState, error)
	Close() error
}

Transport is the semantic boundary implemented by each native stream client adapter. It intentionally contains queue operations rather than a union of arbitrary datastore commands.