dedup

Documentation

Index

• type Deduplicator
  • func NewDeduplicator(config *DeduplicatorConfig) *Deduplicator
  • func (d *Deduplicator) Run()
  • func (d *Deduplicator) RunForever(secondsToSleepBetweenRuns int)
• type DeduplicatorConfig
• type Deleter
  • func NewDeleter(queue Queue, deleteChannel chan string, wg *sync.WaitGroup) *Deleter
  • func (d *Deleter) SetDeleteChannel(deleteChannel chan string)
  • func (d *Deleter) Start()
• type Puller
  • func NewPuller(queue Queue, state *SharedState, messagesExist bool, maxInflight int, ...) *Puller
  • func (p *Puller) MessagesExist() bool
  • func (p *Puller) Start()
• type Queue
• type QueueMessage
• type Reseter
  • func NewReseter(queue Queue, keepChannel chan string, wg *sync.WaitGroup) *Reseter
  • func (r *Reseter) Start()
• type SharedState
  • func NewSharedState(keepMessages map[string]QueueMessage, deleteMessages map[string]struct{}) *SharedState
  • func (s *SharedState) DeleteMessagesLen() int
  • func (s *SharedState) KeepMessagesLen() int
• type Startable

Types

type Deduplicator

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

func NewDeduplicator

func NewDeduplicator(config *DeduplicatorConfig) *Deduplicator

func (*Deduplicator) Run

func (d *Deduplicator) Run()

func (*Deduplicator) RunForever

func (d *Deduplicator) RunForever(secondsToSleepBetweenRuns int)

type DeduplicatorConfig

type DeduplicatorConfig struct {
	Queue       Queue
	NumWorkers  int
	MaxInflight int
}

type Deleter

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

func NewDeleter

func NewDeleter(queue Queue, deleteChannel chan string, wg *sync.WaitGroup) *Deleter

func (*Deleter) SetDeleteChannel

func (d *Deleter) SetDeleteChannel(deleteChannel chan string)

func (*Deleter) Start

func (d *Deleter) Start()

type Puller

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

func NewPuller

func NewPuller(queue Queue, state *SharedState, messagesExist bool, maxInflight int, wg *sync.WaitGroup) *Puller

func (*Puller) MessagesExist

func (p *Puller) MessagesExist() bool

func (*Puller) Start

func (p *Puller) Start()

type Queue

type Queue interface {
	PullMessagesBatch() ([]QueueMessage, error)
	DeleteMessagesBatch(receiptHandles []string)
	ResetVisibilityBatch(receiptHandles []string)
}

type QueueMessage

type QueueMessage interface {
	UniqueID() string
	MessageID() string
	ReceiptHandle() string
}

type Reseter

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

func NewReseter

func NewReseter(queue Queue, keepChannel chan string, wg *sync.WaitGroup) *Reseter

func (*Reseter) Start

func (r *Reseter) Start()

type SharedState

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

func NewSharedState

func NewSharedState(keepMessages map[string]QueueMessage, deleteMessages map[string]struct{}) *SharedState

func (*SharedState) DeleteMessagesLen

func (s *SharedState) DeleteMessagesLen() int

func (*SharedState) KeepMessagesLen

func (s *SharedState) KeepMessagesLen() int

type Startable

type Startable interface {
	Start()
}