eqldbqueue

Documentation

Overview

Package eqldbqueue provides small durable FIFO queues for EQLDB uploads. Queue cursors are byte offsets; acknowledged history is never rescanned.

Index

• Constants
• type Entry
• type Queue
  • func Default() (*Queue, error)
  • func Open(directory string) (*Queue, error)
  • func (q *Queue) Acknowledge(name string, endOffset int64) error
  • func (q *Queue) Append(name, id string, payload any) error
  • func (q *Queue) Batch(name string, maximum int) ([]Entry, error)
  • func (q *Queue) DiscardBefore(cutoff time.Time) error
  • func (q *Queue) Path(name string) string
  • func (q *Queue) RecentIDs(name string, tailBytes int64) (map[string]struct{}, error)

Constants

const (
	PlaneOfSky = "plane-of-sky"
	Kills      = "kills"
	Drops      = "drops"
)

Types

type Entry

type Entry struct {
	ID      string          `json:"id"`
	Payload json.RawMessage `json:"payload"`

	EndOffset int64 `json:"-"`
}

type Queue

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

func Default

func Default() (*Queue, error)

func Open

func Open(directory string) (*Queue, error)

func (*Queue) Acknowledge

func (q *Queue) Acknowledge(name string, endOffset int64) error

func (*Queue) Append

func (q *Queue) Append(name, id string, payload any) error

func (*Queue) Batch

func (q *Queue) Batch(name string, maximum int) ([]Entry, error)

func (*Queue) DiscardBefore

func (q *Queue) DiscardBefore(cutoff time.Time) error

DiscardBefore removes pending launch-data entries older than cutoff and compacts the queues so their byte cursors restart at zero.

func (*Queue) Path

func (q *Queue) Path(name string) string

func (*Queue) RecentIDs

func (q *Queue) RecentIDs(name string, tailBytes int64) (map[string]struct{}, error)

RecentIDs returns IDs from the tail of a queue. Producers use this bounded window to avoid re-enqueueing lines after a crash between queue append and logfile-checkpoint persistence.