queue

Documentation

Index

• Constants
• Variables
• type AssumeRole
• type AssumeRoleResult
• type Configuration
• type Credentials
• type MessageHandler
  • func NewMessageHandler(repositories git.Repositories, actionSpecs conf.ActionSpecifications, ...) MessageHandler
• type Poller
  • func NewPoller(workerPool worker_pool.WorkerPool, queueProvider SQSProvider, ...) Poller
• type Processor
  • func NewProcessor(conf *conf.Configuration) Processor
• type Properties
  • func (p Properties) ExpireTimeMillis() int64
  • func (p Properties) Region() string
  • func (p Properties) Url() string
• type SQSClient
• type SQSProvider
  • func NewSqsProvider(queueProperties Properties) (SQSProvider, error)

Constants

const (
	CustomActionType = "custom"
	HttpActionType   = "http"
)

Variables

var UserAgentHeader string

Types

type AssumeRole

type AssumeRole struct {
	Id  string `json:"id,omitempty"`
	Arn string `json:"arn,omitempty"`
}

type AssumeRoleResult

type AssumeRoleResult struct {
	Credentials Credentials `json:"credentials,omitempty"`
	AssumeRole  AssumeRole  `json:"assumeRole,omitempty"`
}

type Configuration

type Configuration struct {
	SuccessRefreshPeriodInSeconds int64  `json:"credentialSuccessRefreshPeriod,omitempty"`
	ErrorRefreshPeriodInSeconds   int64  `json:"credentialErrorRefreshPeriod,omitempty"`
	Region                        string `json:"region,omitempty"`
	Url                           string `json:"queueUrl,omitempty"`
}

type Credentials

type Credentials struct {
	AccessKeyId      string `json:"accessKeyId,omitempty"`
	SecretAccessKey  string `json:"secretAccessKey,omitempty"`
	SessionToken     string `json:"sessionToken,omitempty"`
	ExpireTimeMillis int64  `json:"expireTimeMillis,omitempty"`
}

type MessageHandler

type MessageHandler interface {
	Handle(message sqs.Message) (*runbook.ActionResultPayload, error)
}

func NewMessageHandler

func NewMessageHandler(repositories git.Repositories, actionSpecs conf.ActionSpecifications, actionLoggers map[string]io.Writer) MessageHandler

type Poller

type Poller interface {
	Processor
	RefreshClient(assumeRoleResult AssumeRoleResult) error
	QueueProvider() SQSProvider
}

func NewPoller

func NewPoller(workerPool worker_pool.WorkerPool,
	queueProvider SQSProvider,
	messageHandler MessageHandler,
	conf *conf.Configuration,
	ownerId string) Poller

type Processor

type Processor interface {
	Start() error
	Stop() error
}

func NewProcessor

func NewProcessor(conf *conf.Configuration) Processor

type Properties

type Properties struct {
	AssumeRoleResult AssumeRoleResult `json:"assumeRoleResult,omitempty"`
	Configuration    Configuration    `json:"queueConfiguration,omitempty"`
}

func (Properties) ExpireTimeMillis

func (p Properties) ExpireTimeMillis() int64

func (Properties) Region

func (p Properties) Region() string

func (Properties) Url

func (p Properties) Url() string

type SQSClient

type SQSClient interface {
	ChangeMessageVisibility(input *sqs.ChangeMessageVisibilityInput) (*sqs.ChangeMessageVisibilityOutput, error)
	DeleteMessage(input *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)
	ReceiveMessage(input *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)
}

type SQSProvider

type SQSProvider interface {
	ChangeMessageVisibility(message *sqs.Message, visibilityTimeout int64) error
	DeleteMessage(message *sqs.Message) error
	ReceiveMessage(numOfMessage int64, visibilityTimeout int64) ([]*sqs.Message, error)

	RefreshClient(assumeRoleResult AssumeRoleResult) error
	Properties() Properties
	IsTokenExpired() bool
}

func NewSqsProvider

func NewSqsProvider(queueProperties Properties) (SQSProvider, error)