queue

Sep 28, 2024

Changes in this version

+ type NoopQueue struct

+ func NewNoopQueue() *NoopQueue

+ func (q *NoopQueue) Ack(m *Message) error

+ func (q *NoopQueue) BatchEnqueue(msg []Message) error

+ func (q *NoopQueue) Enqueue(m *Message) error

+ func (q *NoopQueue) Receive(ch chan Message) error

Aug 14, 2022

Changes in this version

+ const SQSDefaultMaxMessages

+ const SQSDefaultWaitTime

+ const SQSQueueAttributeTrace

+ const SQSTypeString

+ func BatchMessages(messages []Message, chunkSize int) [][]Message

+ func DefaultWorkerBackoffFunc(count int)

+ type Acker interface

+ Ack func(*Message) error

+ type BackoffFunc func(count int)

+ type BatchEnqueuer interface

+ BatchEnqueue func([]Message) error

+ type ChannelQueue struct

+ Input chan Message

+ func NewChannelQueue(input chan Message) *ChannelQueue

+ func (q *ChannelQueue) Ack(m *Message) error

+ func (q *ChannelQueue) BatchEnqueue(msg []Message) error

+ func (q *ChannelQueue) Enqueue(m *Message) error

+ func (q *ChannelQueue) Receive(ch chan Message, done chan bool) error

+ type Enqueuer interface

+ Enqueue func(*Message) error

+ type HandlerFunc func(Message) error

+ type MemoryQueue struct

+ Items []Message

+ func NewMemoryQueue() *MemoryQueue

+ func (q *MemoryQueue) Ack(m *Message) error

+ func (q *MemoryQueue) BatchEnqueue(msg []Message) error

+ func (q *MemoryQueue) Enqueue(m *Message) error

+ func (q *MemoryQueue) Receive(ch chan Message) error

+ type Message struct

+ GroupID string

+ Handle string

+ ID string

+ Payload string

+ Trace string

+ func BuildMessageFromSQSMessage(r *events.SQSMessage) *Message

+ type Queue interface

+ func New(uri string) (Queue, error)

+ type Receiver interface

+ Receive func(chan Message) error

+ type ReceivingAcker interface

+ type RedisPoolQueue struct

+ Topic string

+ func NewRedisPoolQueue(pool *redigo.Pool, topic string) *RedisPoolQueue

+ func (q *RedisPoolQueue) Ack(_ *Message) error

+ func (q *RedisPoolQueue) BatchEnqueue(msgs []Message) error

+ func (q *RedisPoolQueue) Enqueue(m *Message) error

+ func (q *RedisPoolQueue) Receive(ch chan Message) error

+ func (q *RedisPoolQueue) Send(m Message) error

+ func (q *RedisPoolQueue) SendBatch(messages []Message) error

+ type RedisQueue struct

+ Topic string

+ func NewRedisQueue(conn redis.Conn, topic string) *RedisQueue

+ func (q *RedisQueue) Ack(_ *Message) error

+ func (q *RedisQueue) BatchEnqueue(msgs []Message) error

+ func (q *RedisQueue) Enqueue(m *Message) error

+ func (q *RedisQueue) Receive(ch chan<- Message) error

+ func (q *RedisQueue) Send(m Message) error

+ func (q *RedisQueue) SendBatch(messages []Message) error

+ type Runner interface

+ Start func() error

+ Stop func()

+ type SQSQueue struct

+ FIFO bool

+ MaxMessages int64

+ URL string

+ WaitTime int64

+ func NewSQSFIFOQueue(url string) SQSQueue

+ func NewSQSQueue(url string) SQSQueue

+ func (s SQSQueue) Ack(m *Message) error

+ func (s SQSQueue) BatchEnqueue(messages []Message) error

+ func (s SQSQueue) Enqueue(m *Message) error

+ func (s SQSQueue) Receive(messages chan Message) error

+ func (s SQSQueue) Send(m Message) error

+ func (s SQSQueue) SendBatch(messages []Message) error

+ type Worker struct

+ BackoffFunc BackoffFunc

+ ChannelWrite chan Message

+ HandleFunc HandlerFunc

+ Name string

+ Queue ReceivingAcker

+ func NewWorker(name string, r ReceivingAcker, h HandlerFunc) *Worker

+ func (w *Worker) Start() error

+ func (w *Worker) Stop()

+ type Writer interface