sqssrv

README

go-rscsrv-sqs

The go-rscsrv-sqs is the lab259/go-rscsrv service for the Amazon Simple Queue Service (SQS).

It wraps the logic of dealing with credentials and keeping the session.

Dependencies

It depends on the lab259/go-rscsrv (and its dependencies, of course) itself and the aws/aws-sdk-go library.

Installation

First, fetch the library to the repository.

go get github.com/lab259/go-rscsrv-sqs

Usage

The service is designed to be "extended" and not used directly.

srv.go

package mail

import (
	"github.com/lab259/go-rscsrv"
	"github.com/lab259/go-rscsrv-sqs"
)

type MessageSQSService struct {
	sqssrv.SQSService
}

func (service *MessageSQSService) LoadConfiguration() (interface{}, error) {
	var configuration sqssrv.SQSServiceConfiguration

	configurationLoader := rscsrv.NewFileConfigurationLoader("/etc/mail")
	configurationUnmarshaler := &rscsrv.ConfigurationUnmarshalerYaml{}

	config, err := configurationLoader.Load(file)
	if err != nil {
		return err
	}
	return configurationUnmarshaler.Unmarshal(config, dst)
	if err != nil {
		return nil, err
	}
	return configuration, nil
}

example.go

// ...

var mq rscsrsv.MessageSQSService

func init() {
	configuration, err := mq.LoadConfiguration()
	if err != nil {
		panic(err)
	}
	err = mq.ApplyConfiguration(configuration)
	if err != nil {
		panic(err)
	}
	err = mq.Start()
	if err != nil {
		panic(err)
	}
}

// enqueueMessage enqueues a message
func enqueueMessage(message string) {
	output, err := service.SendMessage(&sqs.SendMessageInput{
		MessageBody: aws.String("this is the content of the message"),
	})
	if err != nil {
		panic(err)
	}
	// ... handles the return of the SendMessage
}

// ...

Development

git clone git@github.com:lab259/go-rscsrv-sqs.git # clone the project
cd go-rscsrv-sqs                                  # enter the directory
make dcup                                         # start the ElasticMQ
go mod download                                   # download the dependencies
make test                                         # run the tests

Documentation

Index

• Constants
• type CredentialsFromStruct
  • func NewCredentialsFromStruct(credentials *SQSServiceConfiguration) *CredentialsFromStruct
  • func (*CredentialsFromStruct) IsExpired() bool
  • func (c *CredentialsFromStruct) Retrieve() (credentials.Value, error)
• type SQSService
  • func (service *SQSService) ApplyConfiguration(configuration interface{}) error
  • func (service *SQSService) DeleteMessage(input *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)
  • func (service *SQSService) DeleteMessageBatch(input *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error)
  • func (service *SQSService) DeleteMessageBatchWithContext(ctx context.Context, input *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error)
  • func (service *SQSService) DeleteMessageWithContext(ctx context.Context, input *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)
  • func (service *SQSService) LoadConfiguration() (interface{}, error)
  • func (service *SQSService) PurgeQueue(input *sqs.PurgeQueueInput) (*sqs.PurgeQueueOutput, error)
  • func (service *SQSService) ReceiveMessage(input *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)
  • func (service *SQSService) ReceiveMessageWithContext(ctx context.Context, input *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)
  • func (service *SQSService) Restart() error
  • func (service *SQSService) RunWithSQS(handler func(client *sqs.SQS) error) error
  • func (service *SQSService) SendMessage(input *sqs.SendMessageInput) (*sqs.SendMessageOutput, error)
  • func (service *SQSService) SendMessageBatch(input *sqs.SendMessageBatchInput) (*sqs.SendMessageBatchOutput, error)
  • func (service *SQSService) SendMessageBatchWithContext(ctx context.Context, input *sqs.SendMessageBatchInput) (*sqs.SendMessageBatchOutput, error)
  • func (service *SQSService) SendMessageWithContext(ctx context.Context, input *sqs.SendMessageInput) (*sqs.SendMessageOutput, error)
  • func (service *SQSService) Start() error
  • func (service *SQSService) Stop() error
• type SQSServiceCollector
  • func NewSQSServiceCollector(opts *SQSServiceCollectorOpts) *SQSServiceCollector
  • func (collector *SQSServiceCollector) Collect(metrics chan<- prometheus.Metric)
  • func (collector *SQSServiceCollector) Describe(descs chan<- *prometheus.Desc)
• type SQSServiceCollectorOpts
• type SQSServiceConfiguration

Constants

const (
	MessageMetricMethodSendMessage        string = "SendMessage"
	MessageMetricMethodSendMessageBatch   string = "SendMessageBatch"
	MessageMetricMethodDeleteMessage      string = "DeleteMessage"
	MessageMetricMethodDeleteMessageBatch string = "DeleteMessageBatch"
	MessageMetricMethodReceiveMessage     string = "ReceiveMessage"
)

Types

type CredentialsFromStruct

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

CredentialsFromStruct define credentials from sqs configuration

func NewCredentialsFromStruct

func NewCredentialsFromStruct(credentials *SQSServiceConfiguration) *CredentialsFromStruct

NewCredentialsFromStruct is the configuration for the `SQS`

func (*CredentialsFromStruct) IsExpired

func (*CredentialsFromStruct) IsExpired() bool

IsExpired return if credentials is expired

func (*CredentialsFromStruct) Retrieve

func (c *CredentialsFromStruct) Retrieve() (credentials.Value, error)

Retrieve return the AWS credentials

type SQSService

type SQSService struct {
	Configuration SQSServiceConfiguration
	Collector     *SQSServiceCollector
	// contains filtered or unexported fields
}

SQSService is the service which manages a service queue on the AWS.

func (*SQSService) ApplyConfiguration

func (service *SQSService) ApplyConfiguration(configuration interface{}) error

ApplyConfiguration applies a given configuration to the service.

func (*SQSService) DeleteMessage

func (service *SQSService) DeleteMessage(input *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)

DeleteMessage is a wrapper for the `sqs.SQS.DeleteMessage`.

func (*SQSService) DeleteMessageBatch

func (service *SQSService) DeleteMessageBatch(input *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error)

DeleteMessageBatch is a wrapper for the `sqs.SQS.DeleteMessageBatch`.

func (*SQSService) DeleteMessageBatchWithContext

func (service *SQSService) DeleteMessageBatchWithContext(ctx context.Context, input *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error)

DeleteMessageBatchWithContext is a wrapper for the `sqs.SQS.DeleteMessageBatchWithContext`.

func (*SQSService) DeleteMessageWithContext

func (service *SQSService) DeleteMessageWithContext(ctx context.Context, input *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)

DeleteMessageWithContext is a wrapper for the `sqs.SQS.DeleteMessageWithContext`.

func (*SQSService) LoadConfiguration

func (service *SQSService) LoadConfiguration() (interface{}, error)

LoadConfiguration returns

func (*SQSService) PurgeQueue

func (service *SQSService) PurgeQueue(input *sqs.PurgeQueueInput) (*sqs.PurgeQueueOutput, error)

PurgeQueue is a wrapper for the `sqs.SQS.PurgeQueue`.

func (*SQSService) ReceiveMessage

func (service *SQSService) ReceiveMessage(input *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)

ReceiveMessage is a wrapper for the `sqs.SQS.ReceiveMessage`.

func (*SQSService) ReceiveMessageWithContext

func (service *SQSService) ReceiveMessageWithContext(ctx context.Context, input *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)

ReceiveMessageWithContext is a wrapper for the `sqs.SQS.ReceiveMessageWithContext`.

func (*SQSService) Restart

func (service *SQSService) Restart() error

Restart stops and then starts the service again.

func (*SQSService) RunWithSQS

func (service *SQSService) RunWithSQS(handler func(client *sqs.SQS) error) error

RunWithSQS runs a handler passing the reference of a `sqs.SQS` client.

func (*SQSService) SendMessage

func (service *SQSService) SendMessage(input *sqs.SendMessageInput) (*sqs.SendMessageOutput, error)

SendMessage is a wrapper for the `sqs.SQS.SendMessage`.

func (*SQSService) SendMessageBatch

func (service *SQSService) SendMessageBatch(input *sqs.SendMessageBatchInput) (*sqs.SendMessageBatchOutput, error)

SendMessageBatch is a wrapper for the `sqs.SQS.SendMessageBatch`.

func (*SQSService) SendMessageBatchWithContext

func (service *SQSService) SendMessageBatchWithContext(ctx context.Context, input *sqs.SendMessageBatchInput) (*sqs.SendMessageBatchOutput, error)

SendMessageBatchWithContext is a wrapper for the `sqs.SQS.SendMessageBatchWithContext`.

func (*SQSService) SendMessageWithContext

func (service *SQSService) SendMessageWithContext(ctx context.Context, input *sqs.SendMessageInput) (*sqs.SendMessageOutput, error)

SendMessageWithContext is a wrapper for the `sqs.SQS.SendMessage`.

func (*SQSService) Start

func (service *SQSService) Start() error

Start starts the service pool.

func (*SQSService) Stop

func (service *SQSService) Stop() error

Stop erases the aws client reference.

type SQSServiceCollector

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

func NewSQSServiceCollector

func NewSQSServiceCollector(opts *SQSServiceCollectorOpts) *SQSServiceCollector

func (*SQSServiceCollector) Collect

func (collector *SQSServiceCollector) Collect(metrics chan<- prometheus.Metric)

func (*SQSServiceCollector) Describe

func (collector *SQSServiceCollector) Describe(descs chan<- *prometheus.Desc)

type SQSServiceCollectorOpts

type SQSServiceCollectorOpts struct {
	Prefix string
}

type SQSServiceConfiguration

type SQSServiceConfiguration struct {
	QUrl            string `yaml:"q_url"`
	Region          string `yaml:"region"`
	Endpoint        string `yaml:"endpoint"`
	Key             string `yaml:"key"`
	Secret          string `yaml:"secret"`
	CollectorPrefix string `yaml:"collector_prefix"`
}

SQSServiceConfiguration is the configuration for the `SQS`