sqs

Documentation

Index

• Constants
• func FromPipeline(pipe *pipeline.Pipeline, log logger.Logger, cfg cfgPlugin.Configurer, ...) (*consumer, error)
• func NewSQSConsumer(configKey string, log logger.Logger, cfg cfgPlugin.Configurer, ...) (*consumer, error)
• type Config
  • func (c *Config) InitDefault()
• type GlobalCfg
  • func (c *GlobalCfg) InitDefault()
• type Item
  • func (i *Item) Ack() error
  • func (i *Item) Body() []byte
  • func (i *Item) Context() ([]byte, error)
  • func (i *Item) ID() string
  • func (i *Item) Nack() error
  • func (i *Item) Priority() int64
  • func (i *Item) Requeue(headers map[string][]string, delay int64) error
• type Options
  • func (o *Options) DelayDuration() time.Duration
• type Plugin
  • func (p *Plugin) Available()
  • func (p *Plugin) FromPipeline(pipe *pipeline.Pipeline, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error)
  • func (p *Plugin) Init(log logger.Logger, cfg config.Configurer) error
  • func (p *Plugin) JobsConstruct(configKey string, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error)
  • func (p *Plugin) Name() string

Constants

const (
	StringType              string = "String"
	NumberType              string = "Number"
	BinaryType              string = "Binary"
	ApproximateReceiveCount string = "ApproximateReceiveCount"
)

const (
	// All - get all message attribute names
	All string = "All"

	// NonExistentQueue AWS error code
	NonExistentQueue string = "AWS.SimpleQueueService.NonExistentQueue"
)

Functions

func FromPipeline

func FromPipeline(pipe *pipeline.Pipeline, log logger.Logger, cfg cfgPlugin.Configurer, e events.Handler, pq priorityqueue.Queue) (*consumer, error)

func NewSQSConsumer

func NewSQSConsumer(configKey string, log logger.Logger, cfg cfgPlugin.Configurer, e events.Handler, pq priorityqueue.Queue) (*consumer, error)

Types

type Config

type Config struct {
	// The duration (in seconds) that the received messages are hidden from subsequent
	// retrieve requests after being retrieved by a ReceiveMessage request.
	VisibilityTimeout int32 `mapstructure:"visibility_timeout"`
	// The duration (in seconds) for which the call waits for a message to arrive
	// in the queue before returning. If a message is available, the call returns
	// sooner than WaitTimeSeconds. If no messages are available and the wait time
	// expires, the call returns successfully with an empty list of messages.
	WaitTimeSeconds int32 `mapstructure:"wait_time_seconds"`
	// Prefetch is the maximum number of messages to return. Amazon SQS never returns more messages
	// than this value (however, fewer messages might be returned). Valid values: 1 to
	// 10. Default: 1.
	Prefetch int32 `mapstructure:"prefetch"`
	// The name of the new queue. The following limits apply to this name:
	//
	// * A queue
	// name can have up to 80 characters.
	//
	// * Valid values: alphanumeric characters,
	// hyphens (-), and underscores (_).
	//
	// * A FIFO queue name must end with the .fifo
	// suffix.
	//
	// Queue URLs and names are case-sensitive.
	//
	// This member is required.
	Queue *string `mapstructure:"queue"`

	// A map of attributes with their corresponding values. The following lists the
	// names, descriptions, and values of the special request parameters that the
	// CreateQueue action uses.
	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SetQueueAttributes.html
	Attributes map[string]string `mapstructure:"attributes"`

	// From amazon docs:
	// Add cost allocation tags to the specified Amazon SQS queue. For an overview, see
	// Tagging Your Amazon SQS Queues
	// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-tags.html)
	// in the Amazon SQS Developer Guide. When you use queue tags, keep the following
	// guidelines in mind:
	//
	// * Adding more than 50 tags to a queue isn't recommended.
	//
	// *
	// Tags don't have any semantic meaning. Amazon SQS interprets tags as character
	// strings.
	//
	// * Tags are case-sensitive.
	//
	// * A new tag with a key identical to that
	// of an existing tag overwrites the existing tag.
	//
	// For a full list of tag
	// restrictions, see Quotas related to queues
	// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-limits.html#limits-queues)
	// in the Amazon SQS Developer Guide. To be able to tag a queue on creation, you
	// must have the sqs:CreateQueue and sqs:TagQueue permissions. Cross-account
	// permissions don't apply to this action. For more information, see Grant
	// cross-account permissions to a role and a user name
	// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-customer-managed-policy-examples.html#grant-cross-account-permissions-to-role-and-user-name)
	// in the Amazon SQS Developer Guide.
	Tags map[string]string `mapstructure:"tags"`
}

Config is used to parse pipeline configuration

func (*Config) InitDefault

func (c *Config) InitDefault()

type GlobalCfg

type GlobalCfg struct {
	Key          string `mapstructure:"key"`
	Secret       string `mapstructure:"secret"`
	Region       string `mapstructure:"region"`
	SessionToken string `mapstructure:"session_token"`
	Endpoint     string `mapstructure:"endpoint"`
}

func (*GlobalCfg) InitDefault

func (c *GlobalCfg) InitDefault()

type Item

type Item struct {
	// Job contains pluginName of job broker (usually PHP class).
	Job string `json:"job"`

	// Ident is unique identifier of the job, should be provided from outside
	Ident string `json:"id"`

	// Payload is string data (usually JSON) passed to Job broker.
	Payload string `json:"payload"`

	// Headers with key-values pairs
	Headers map[string][]string `json:"headers"`

	// Options contains set of PipelineOptions specific to job execution. Can be empty.
	Options *Options `json:"options,omitempty"`
}

func (*Item) Ack

func (i *Item) Ack() error

func (*Item) Body

func (i *Item) Body() []byte

Body packs job payload into binary payload.

func (*Item) Context

func (i *Item) Context() ([]byte, error)

Context packs job context (job, id) into binary payload. Not used in the sqs, MessageAttributes used instead

func (*Item) ID

func (i *Item) ID() string

func (*Item) Nack

func (i *Item) Nack() error

func (*Item) Priority

func (i *Item) Priority() int64

func (*Item) Requeue

func (i *Item) Requeue(headers map[string][]string, delay int64) error

type Options

type Options struct {
	// Priority is job priority, default - 10
	// pointer to distinguish 0 as a priority and nil as priority not set
	Priority int64 `json:"priority"`

	// Pipeline manually specified pipeline.
	Pipeline string `json:"pipeline,omitempty"`

	// Delay defines time duration to delay execution for. Defaults to none.
	Delay int64 `json:"delay,omitempty"`
	// contains filtered or unexported fields
}

Options carry information about how to handle given job.

func (*Options) DelayDuration

func (o *Options) DelayDuration() time.Duration

DelayDuration returns delay duration in a form of time.Duration.

type Plugin

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

func (*Plugin) Available

func (p *Plugin) Available()

func (*Plugin) FromPipeline

func (p *Plugin) FromPipeline(pipe *pipeline.Pipeline, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error)

func (*Plugin) Init

func (p *Plugin) Init(log logger.Logger, cfg config.Configurer) error

func (*Plugin) JobsConstruct

func (p *Plugin) JobsConstruct(configKey string, e events.Handler, pq priorityqueue.Queue) (jobs.Consumer, error)

func (*Plugin) Name

func (p *Plugin) Name() string