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
- type AddRequest
- type ClaimRequest
- type ClaimResult
- type DeadLetterRequest
- type Delivery
- type FailureMetadata
- type GroupState
- type ReadRequest
- type RequestError
- 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 ¶
MessageAge derives an entry age from a server-generated stream identifier.
func MessageMetadata ¶
MessageMetadata returns validated operational metadata from the bounded job envelope. Malformed and legacy bodies have no trustworthy metadata.
Types ¶
type AckRequest ¶
AckRequest identifies one consumer-group delivery to settle.
func (AckRequest) Validate ¶
func (r AckRequest) Validate() error
Validate checks acknowledgement identity.
type AddRequest ¶
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 ¶
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 ¶
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 ¶
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 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.
Source Files