sqs

Documentation

Index

• Constants
• type Reader
  • func NewReader(log log.Logger, cfg config.SQSReader, maxTimeNoRead *time.Duration, ...) (*Reader, error)
  • func (r *Reader) LastMessageReceived() *time.Time
  • func (r *Reader) SetMessageProcessor(p queue.MessageProcessor)
  • func (r *Reader) StartReading(ctx context.Context) <-chan error
• type Writer
  • func NewWriter(queueARN string, endpoint string, log log.Logger) (*Writer, error)
  • func (w *Writer) Write(body string) error

Constants

const (
	MaxQuantumDelta = 3 // in seconds
)

Types

type Reader

type Reader struct {
	*sync.RWMutex

	Processor queue.MessageProcessor
	// contains filtered or unexported fields
}

func NewReader

func NewReader(log log.Logger, cfg config.SQSReader, maxTimeNoRead *time.Duration, processor queue.MessageProcessor) (*Reader, error)

NewReader creates a new Reader with the given processor, queueARN and config.

func (*Reader) LastMessageReceived

func (r *Reader) LastMessageReceived() *time.Time

LastMessageReceived returns the time where the last message was received by the Reader. If no message was received so far it returns nil.

func (*Reader) SetMessageProcessor

func (r *Reader) SetMessageProcessor(p queue.MessageProcessor)

SetMessageProcessor sets the queue's message processor. It must be set before calling *Reader.StartReading.

func (*Reader) StartReading

func (r *Reader) StartReading(ctx context.Context) <-chan error

StartReading starts reading messages from the sqs queue. It reads messages only when there are free tokens in the message processor. It will stop reading from the queue when the passed in context is canceled. The caller can use the returned channel to track when the reader stopped reading from the queue and all the messages it is tracking are finished processing.

type Writer

type Writer struct {
	*sync.RWMutex
	// contains filtered or unexported fields
}

Writer writes messages to and AWS SQS queue.

func NewWriter

func NewWriter(queueARN string, endpoint string, log log.Logger) (*Writer, error)

NewWriter creates a new SQS writer to writer to given queue ARN using the passed in endpoint, or the default one if the it is empty.

func (*Writer) Write

func (w *Writer) Write(body string) error