sqs

Documentation

Index

• Constants
• type AttributesResponse
  • func NewAttributesResponse(apiResponse map[string]*string) AttributesResponse
• type FailedResults
• type Message
  • func NewMessage(msg *SDK.Message) *Message
  • func (m *Message) Body() string
  • func (m *Message) GetMessageID() *string
  • func (m *Message) GetReceiptHandle() *string
  • func (m *Message) String() string
• type Queue
  • func NewQueue(svc *SQS, name string, url string) *Queue
  • func (q *Queue) AddDeleteList(msg interface{})
  • func (q *Queue) AddMessage(message string)
  • func (q *Queue) AddMessageJSONMarshal(message interface{}) error
  • func (q *Queue) AddMessageMap(message map[string]interface{}) error
  • func (q *Queue) AddMessageWithGroupID(message string, messageGroupID string)
  • func (q *Queue) AutoDelete(b bool)
  • func (q *Queue) ChangeMessageVisibility(msg *Message, timeoutInSeconds int) error
  • func (q *Queue) ClearFailedResults()
  • func (q *Queue) CountMessage() (visible int, invisible int, err error)
  • func (q *Queue) DeleteListItems() error
  • func (q *Queue) DeleteMessage(msg *Message) error
  • func (q *Queue) DeleteMessageWithReceipt(msgReceipt string) error
  • func (q *Queue) Fetch(num int) ([]*Message, error)
  • func (q *Queue) FetchBody(num int) []string
  • func (q *Queue) FetchBodyOne() string
  • func (q *Queue) FetchOne() (*Message, error)
  • func (q *Queue) GetAttributes() (AttributesResponse, error)
  • func (q *Queue) GetFailedResults() FailedResults
  • func (q *Queue) Purge() error
  • func (q *Queue) Send() error
  • func (q *Queue) SetExpire(sec int)
  • func (q *Queue) SetWaitTimeSeconds(sec int)
• type SQS
  • func New(conf config.Config) (*SQS, error)
  • func (svc *SQS) CreateQueue(in *SDK.CreateQueueInput) error
  • func (svc *SQS) CreateQueueWithName(name string) error
  • func (svc *SQS) DeleteQueue(name string) error
  • func (svc *SQS) Errorf(format string, v ...interface{})
  • func (svc *SQS) GetQueue(name string) (*Queue, error)
  • func (svc *SQS) GetQueueAttributes(url string, attributes ...string) (AttributesResponse, error)
  • func (svc *SQS) Infof(format string, v ...interface{})
  • func (svc *SQS) IsExistQueue(name string) (bool, error)
  • func (svc *SQS) ListAllQueues() ([]string, error)
  • func (svc *SQS) ListQueues(prefix string) ([]string, error)
  • func (svc *SQS) SetLogger(logger log.Logger)
  • func (svc *SQS) SetPrefix(prefix string)

Constants

const (
	AttributeAll                                   = "All"
	AttributeApproximateNumberOfMessages           = "ApproximateNumberOfMessages"
	AttributeApproximateNumberOfMessagesDelayed    = "ApproximateNumberOfMessagesDelayed"
	AttributeApproximateNumberOfMessagesNotVisible = "ApproximateNumberOfMessagesNotVisible"
	AttributeCreatedTimestamp                      = "CreatedTimestamp"
	AttributeDelaySeconds                          = "DelaySeconds"
	AttributeLastModifiedTimestamp                 = "LastModifiedTimestamp"
	AttributeMaximumMessageSize                    = "MaximumMessageSize"
	AttributeMessageRetentionPeriod                = "MessageRetentionPeriod"
	AttributePolicy                                = "Policy"
	AttributeQueueArn                              = "QueueArn"
	AttributeReceiveMessageWaitTimeSeconds         = "ReceiveMessageWaitTimeSeconds"
	AttributeRedrivePolicy                         = "RedrivePolicy"
	AttributeVisibilityTimeout                     = "VisibilityTimeout"
	AttributeKmsMasterKeyId                        = "KmsMasterKeyId" // nolint
	AttributeKmsDataKeyReusePeriodSeconds          = "KmsDataKeyReusePeriodSeconds"
	AttributeFifoQueue                             = "FifoQueue"
	AttributeContentBasedDeduplication             = "ContentBasedDeduplication"

	AttributeRedrivePolicyDeadLetterTargetArn = "deadLetterTargetArn"
	AttributeRedrivePolicyMaxReceiveCount     = "maxReceiveCount"
)

Attribute names for SQS. ref: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_GetQueueAttributes.html

Types

type AttributesResponse

type AttributesResponse struct {
	ApproximateNumberOfMessages           int
	ApproximateNumberOfMessagesDelayed    int
	ApproximateNumberOfMessagesNotVisible int
	CreatedTimestamp                      int
	DelaySeconds                          int
	LastModifiedTimestamp                 int
	MaximumMessageSize                    int
	MessageRetentionPeriod                int
	QueueArn                              string
	ReceiveMessageWaitTimeSeconds         int
	RedrivePolicy                         string
	VisibilityTimeout                     int
}

AttributesResponse contains attributes from GetQueueAttributes.

func NewAttributesResponse

func NewAttributesResponse(apiResponse map[string]*string) AttributesResponse

type FailedResults

type FailedResults struct {
	Send   []*SDK.BatchResultErrorEntry
	Delete []*SDK.BatchResultErrorEntry
}

FailedResults contains failed results of batch request.

type Message

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

Message is SQS Message wrapper struct.

func NewMessage

func NewMessage(msg *SDK.Message) *Message

NewMessage returns initialized *Message.

func (*Message) Body

func (m *Message) Body() string

Body returns message body.

func (*Message) GetMessageID

func (m *Message) GetMessageID() *string

GetMessageID returns pointer of message id.

func (*Message) GetReceiptHandle

func (m *Message) GetReceiptHandle() *string

GetReceiptHandle returns pointer of ReceiptHandle.

func (*Message) String

func (m *Message) String() string

type Queue

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

Queue is SQS Queue wrapper struct.

func NewQueue

func NewQueue(svc *SQS, name string, url string) *Queue

NewQueue returns initialized *Queue.

func (*Queue) AddDeleteList

func (q *Queue) AddDeleteList(msg interface{})

AddDeleteList adds a message to the delete spool.

func (*Queue) AddMessage

func (q *Queue) AddMessage(message string)

AddMessage adds message to the send spool. This assumes a Standard SQS Queue and not a FifoQueue

func (*Queue) AddMessageJSONMarshal

func (q *Queue) AddMessageJSONMarshal(message interface{}) error

AddMessageJSONMarshal adds message to the send pool with encoding json data.

func (*Queue) AddMessageMap

func (q *Queue) AddMessageMap(message map[string]interface{}) error

AddMessageMap adds message to the send pool from map data.

func (*Queue) AddMessageWithGroupID

func (q *Queue) AddMessageWithGroupID(message string, messageGroupID string)

AddMessageWithGroupID adds a message to the send spool but adds the required attributes for a SQS FIFO Queue. This assumes the SQS FIFO Queue has ContentBasedDeduplication enabled.

func (*Queue) AutoDelete

func (q *Queue) AutoDelete(b bool)

AutoDelete sets auto delete flag.

func (*Queue) ChangeMessageVisibility

func (q *Queue) ChangeMessageVisibility(msg *Message, timeoutInSeconds int) error

ChangeMessageVisibility sends the request to AWS api to change visibility of the message.

func (*Queue) ClearFailedResults

func (q *Queue) ClearFailedResults()

ClearFailedResults resets failed results of batch request.

func (*Queue) CountMessage

func (q *Queue) CountMessage() (visible int, invisible int, err error)

CountMessage sends request to AWS api to counts left messages in the Queue.

func (*Queue) DeleteListItems

func (q *Queue) DeleteListItems() error

DeleteListItems executes delete operation in the delete spool.

func (*Queue) DeleteMessage

func (q *Queue) DeleteMessage(msg *Message) error

DeleteMessage sends the request to AWS api to delete the message.

func (*Queue) DeleteMessageWithReceipt

func (q *Queue) DeleteMessageWithReceipt(msgReceipt string) error

DeleteMessageWithReceipt sends the request to AWS api to delete the message.

func (*Queue) Fetch

func (q *Queue) Fetch(num int) ([]*Message, error)

Fetch fetches message list from the queue with limit.

func (*Queue) FetchBody

func (q *Queue) FetchBody(num int) []string

FetchBody fetches only the body of messages. ** cannot handle deletion manually as lack of MessageId and ReceiptHandle **

func (*Queue) FetchBodyOne

func (q *Queue) FetchBodyOne() string

FetchBodyOne fetches the body of a single message. ** cannot handle deletion manually as lack of MessageId and ReceiptHandle **

func (*Queue) FetchOne

func (q *Queue) FetchOne() (*Message, error)

FetchOne fetches a single message.

func (*Queue) GetAttributes

func (q *Queue) GetAttributes() (AttributesResponse, error)

GetAttributes sends request to AWS api to get the queue's attributes. `AttributeNames` will be set as `All`.

func (*Queue) GetFailedResults

func (q *Queue) GetFailedResults() FailedResults

GetFailedResults gets failed results of batch request.

func (*Queue) Purge

func (q *Queue) Purge() error

Purge deletes all messages in the Queue.

func (*Queue) Send

func (q *Queue) Send() error

Send sends messages in the send spool

func (*Queue) SetExpire

func (q *Queue) SetExpire(sec int)

SetExpire sets visibility timeout for message.

func (*Queue) SetWaitTimeSeconds

func (q *Queue) SetWaitTimeSeconds(sec int)

SetWaitTimeSeconds sets wait time timeout for message. Setting this value allows for a long polling workflow.

type SQS

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

SQS has SQS client and Queue list.

func New

func New(conf config.Config) (*SQS, error)

New returns initialized *SQS.

func (*SQS) CreateQueue

func (svc *SQS) CreateQueue(in *SDK.CreateQueueInput) error

CreateQueue creates new SQS Queue.

func (*SQS) CreateQueueWithName

func (svc *SQS) CreateQueueWithName(name string) error

CreateQueueWithName creates new SQS Queue by given name

func (*SQS) DeleteQueue

func (svc *SQS) DeleteQueue(name string) error

DeleteQueue detes the SQS Queue.

func (*SQS) Errorf

func (svc *SQS) Errorf(format string, v ...interface{})

Errorf logging error information.

func (*SQS) GetQueue

func (svc *SQS) GetQueue(name string) (*Queue, error)

GetQueue gets SQS Queue.

func (*SQS) GetQueueAttributes

func (svc *SQS) GetQueueAttributes(url string, attributes ...string) (AttributesResponse, error)

GetQueueAttributes gets the queue's attributes.

func (*SQS) Infof

func (svc *SQS) Infof(format string, v ...interface{})

Infof logging information.

func (*SQS) IsExistQueue

func (svc *SQS) IsExistQueue(name string) (bool, error)

IsExistQueue checks if the Queue already exists or not.

func (*SQS) ListAllQueues

func (svc *SQS) ListAllQueues() ([]string, error)

ListQueues gets all of SQS queue list.

func (*SQS) ListQueues

func (svc *SQS) ListQueues(prefix string) ([]string, error)

ListQueues gets SQS queues list.

func (*SQS) SetLogger

func (svc *SQS) SetLogger(logger log.Logger)

SetLogger sets logger.

func (*SQS) SetPrefix

func (svc *SQS) SetPrefix(prefix string)

SetPrefix sets prefix.