queue

Documentation

Index

• type DeadLetterQueue
  • func NewDLQ(dir string, interval time.Duration, replayFn func(data []byte) error) (*DeadLetterQueue, error)
  • func (d *DeadLetterQueue) Enqueue(batch interface{}) error
  • func (d *DeadLetterQueue) Size() int
  • func (d *DeadLetterQueue) Stop()

Types

type DeadLetterQueue

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

DeadLetterQueue provides disk-based resilience for failed database writes. When a batch insert fails, the data is serialized to JSON and written to disk. A background replay worker periodically attempts to re-insert failed batches.

func NewDLQ

func NewDLQ(dir string, interval time.Duration, replayFn func(data []byte) error) (*DeadLetterQueue, error)

NewDLQ creates a new Dead Letter Queue that stores failed batches in the given directory. replayFn is called with raw JSON bytes during replay — the caller provides the deserialization + insert logic.

func (*DeadLetterQueue) Enqueue

func (d *DeadLetterQueue) Enqueue(batch interface{}) error

Enqueue serializes the given batch to JSON and writes it to disk. This is called when a database batch insert fails.

func (*DeadLetterQueue) Size

func (d *DeadLetterQueue) Size() int

Size returns the number of files currently in the DLQ directory.

func (*DeadLetterQueue) Stop

func (d *DeadLetterQueue) Stop()

Stop gracefully shuts down the replay worker.