queue

Documentation

Overview

Package queue provides functionality for creating and running an engine.

Index

• Variables
• type HandlerFunc
• type InvalidBlockError
  • func (q *InvalidBlockError) Error() string
• type Message
• type MissingHandlerError
  • func (q *MissingHandlerError) Error() string
• type Queue
  • func New(cfg *config.Configuration, hs HandlerFunc) (*Queue, error)
  • func (q *Queue) Close()
  • func (q *Queue) Start(ctx context.Context)

Variables

var (
	// ErrMissingConsumerQueue is returned when the consumer queue name is missing.
	ErrMissingConsumerQueue = errors.New("missing consumer queue name")
	// ErrMissingConsumerQueueURL is returned when the consumer queue url is missing.
	ErrMissingConsumerQueueURL = errors.New("missing consumer queue url")
	// ErrMissingProducerQueue is returned when the producer queue name is missing.
	ErrMissingProducerQueue = errors.New("missing producer queue name")
	// ErrMissingRegion is returned when the queue region is missing.
	ErrMissingRegion = errors.New("missing queue region")
)

var BackoffStrategy = func() backoff.BackOff {
	return &backoff.ExponentialBackOff{
		Clock:               backoff.SystemClock,
		InitialInterval:     5 * time.Second,
		MaxInterval:         10 * time.Second,
		MaxElapsedTime:      300 * time.Second,
		Multiplier:          backoff.DefaultMultiplier,
		RandomizationFactor: backoff.DefaultRandomizationFactor,
	}
}

BackoffStrategy is the backoff strategy used when attempting retryable errors.

var ErrHandler = func(ctx context.Context, event string, err error) {
	log.Error(ctx, event, err)
}

ErrHandler is the handler function applied to an error.

Types

type HandlerFunc

type HandlerFunc func(ctx context.Context, cfg config.Configuration, msg *message.MessageSQS) error

HandlerFunc represents a function that is applied to a consumed message.

type InvalidBlockError

type InvalidBlockError struct {
	MessageID string
}

InvalidBlockError is an error implementation that includes a consumed message ID.

func (*InvalidBlockError) Error

func (q *InvalidBlockError) Error() string

type Message

type Message struct {
	Artifacts []string
	Bucket    string
	ID        string `json:"-"`
	Service   string
	Type      string
}

Message represents a message that has been consumed.

type MissingHandlerError

type MissingHandlerError struct {
	MessageType string
}

MissingHandlerError is an error implementation that includes a consumed message type.

func (*MissingHandlerError) Error

func (q *MissingHandlerError) Error() string

type Queue

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

Queue represents a Queue.

func New

func New(cfg *config.Configuration, hs HandlerFunc) (*Queue, error)

New returns a new queue.

func (*Queue) Close

func (q *Queue) Close()

Close ssqs queue

func (*Queue) Start

func (q *Queue) Start(ctx context.Context)

Start starts the queue consumer and applies a given handler function to each message that is consumed. Once the message has successfully been handled, we attempt to write the result of the handler function to an outbound queue. If the result is written successfully, the message that was originally consumed is removed from the queue.