init

Author: Quest Henkart

LICENSE.txt

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) [2020] [Quest Henkart]
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,72 @@
+# gosqs
+
+GoSQS serves as the messaging interface between AWS-SQS and AWS-SNS services. If there is an feature you would like implemented, please make a Pull Request
+
+Please refer to the examples to see how to interact with this library. Please make contributions, and post any issues.
+
+Take a look at the Dead Letter Queue and Naming your Queue as they are important for this design pattern
+
+## TODO
+- create better documentation
+- implement internal testing package to avoid dependency
+- create a controller that scales the worker pool up automatically depending on the message count and then kills the workers to avoid costs
+
+## Scaling the Consumers
+Each SQS consumer is ready to be scaled up as well as scaled out.
+
+* Scaling Out: You can scale out by creating additional instances with no extra configuration required. The library and SQS is designed for multiple workers reaching into the same pool
+
+* Scaling Up: Each consumer has a configuration variable `config.WorkerPool`. The default is set to `30`, that means there are 30 goroutines checking for messages at any given time. You can increase the amount of active threads simply by adjusting that number. Make sure to monitor CPU usage to find the right count for your application. For a local or dev environment. Reduce this number to 1 to save battery
+
+## Configuring SNS
+configuring SNS is easy, simply login to the AWS-console, navigate to SNS. Click on Topics on the sidebar and "Create New Topic". Fill in the name and display name.
+* make sure to set the topic delivery policy to exponential back off
+
+## Configuring SQS
+1. Navigate to aws-sqs
+2. Choose a queue Name and click on Standard Queue
+3. Click configure, apply optional configurations
+4. Hit Create Queue
+5. in the main page, click on the newly created queue
+6. Click on Queue Actions and at the bottom hit "Subscribe Queue to SNS Topic"
+7. Select the SNS topic from the dropdown provided
+
+## Naming your Queue
+The naming convention for queues supported by this library follow the following syntax
+
+<env>-<name>
+
+## SQS Configurations  
+
+### Default Visibility Timeout  
+The default visibility timeout is responsible for managing how often a single message gets received by a consumer. A message remains in the queue until it is deleted, and any receipt of the message that does not end in deletion before the Visibility Timeout is hit is considered a "failure". As a result, the Default Visibility Timeout should exceed the maximum possible amount of time it would take for a handler to process and delete a message.  
+
+The currently set default is 30 seconds
+
+*note* The visibility timeout is extended for an individual message, for a maximum of 3 x the visibility timeout
+
+### Message Retention Period
+The # of days that the Queue will hold on to an unconsumed message before deleting it. Since we will always be consuming, this value is not important, the default is 4 days
+
+### Receive Message Wait Time
+The amount of time that the request will hang before returning 0 messages. This field is important as it allows us to use long-polling instead of short-polling. The default is 0 and it *should be set to the max 20 seconds to save on processing and cost*. AWS recommends using long polling over short polling
+
+### Custom Attributes
+You can add custom attributes to your SQS implementation. These are fields that exist outside of the payload body. A common practice is to include a correlationId or some sort of trackingId to track a message
+
+
+### DEAD LETTER QUEUE CONFIGURATION
+The following settings activate an automatic reroute to the DLQ upon repetetive failure of message processing.
+* Redrive Policy must be checked
+* The Dead Letter Queue name must be provided (A DLQ is just a normal SQS)
+* Maximum Receives reflects the amount of times a message is received, but not deleted before it is requeued into the DLQ  
+* *Including a DLQ is an absolute must, do not run a system without it our you will be vulnerable to Poison-Pill attacks*
+
+## Consumer Configuration
+
+### Custom Middleware
+You can add custom middleware to your consumer. These will run using the adapter method before each handler is called. You can include a logger or modify the context etc
+
+## Testing
+You can set up a local SNS/SQS emulator using https://github.com/p4tin/goaws. Contributions have been added to this emulator specifically to support this library
+Tests also require this to be running, I will eventually set up a ci environment that runs the emulator in a container and runs the tests

adapters.go

@@ -0,0 +1,63 @@
+package gosqs
+
+import (
+	"context"
+)
+
+const (
+	dispatcherKey = contextKey("dispatcher")
+)
+
+type contextKey string
+
+// Handler provides a standardized handler method, this is the required function composition for event handlers
+type Handler func(context.Context, Message) error
+
+// Adapter implements adapters in the context
+type Adapter func(Handler) Handler
+
+// WithRecovery is an adapter that logs a Panic error and recovers the service from a failed state
+func WithRecovery(recovery func()) Adapter {
+	return func(fn Handler) Handler {
+		return func(ctx context.Context, m Message) error {
+			defer recovery()
+
+			return fn(ctx, m)
+		}
+	}
+}
+
+// WithMiddleware add middleware to the consumer service
+func WithMiddleware(f func(ctx context.Context, m Message) error) Adapter {
+	return func(fn Handler) Handler {
+		return func(ctx context.Context, m Message) error {
+			f(ctx, m)
+
+			return fn(ctx, m)
+		}
+	}
+}
+
+// WithDispatcher sets an adapter to support sending async messages
+func WithDispatcher(ctx context.Context, pub Publisher) context.Context {
+	return context.WithValue(ctx, dispatcherKey, pub)
+}
+
+// Dispatcher retrieves the sqs dispatcher from the context for sending messeges
+func Dispatcher(ctx context.Context) (Publisher, error) {
+	if p, ok := ctx.Value(dispatcherKey).(Publisher); ok {
+		return p, nil
+	}
+
+	return nil, ErrUndefinedPublisher
+}
+
+// MustDispatcher retrieves the sqs dispatcher from the context for sending messeges or panics if
+// the Dispatcher does not exist in the context
+func MustDispatcher(ctx context.Context) Publisher {
+	if p, ok := ctx.Value(dispatcherKey).(Publisher); ok {
+		return p
+	}
+
+	panic(ErrUndefinedPublisher.Error())
+}

config.go

@@ -0,0 +1,132 @@
+package gosqs
+
+import (
+	"strconv"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/client"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/aws/session"
+)
+
+// Config defines the gosqs configuration
+type Config struct {
+	// private key to access aws
+	Key string
+	// secret to access aws
+	Secret string
+	// region for aws and used for determining the topic ARN
+	Region string
+	// provided automatically by aws, but must be set for emulators or local testing
+	Hostname string
+	// account ID of the aws account, used for determining the topic ARN
+	AWSAccountID string
+	// environment name, used for determinig the topic ARN
+	Env string
+	// prefix of the topic, this is set as a prefix to the environment
+	TopicPrefix string
+	// optional address of the topic, if this is not provided it will be created using other variables
+	TopicARN string
+	// optional address of queue, if this is not provided it will be retrieved during setup
+	QueueURL string
+	// used to extend the allowed processing time of a message
+	VisibilityTimeout int
+	// used to determine how many attempts exponential backoff should use before logging an error
+	RetryCount int
+	// defines the total amount of goroutines that can be run by the consumer
+	WorkerPool int
+	// defines the total number of processing extensions that occur. Each proccessing extension will double the
+	// visibilitytimeout counter, ensuring the handler has more time to process the message. Default is 2 extensions (1m30s processing time)
+	// set to 0 to turn off extension processing
+	ExtensionLimit *int
+
+	// Add custom attributes to the message. This might be a correlationId or client meta information
+	// custom attributes will be viewable on the sqs dashboard as meta data
+	Attributes []customAttribute
+
+	// Add a custom logger, the default will be log.Println
+	Logger Logger
+}
+
+// customAttribute add custom attributes to SNS and SQS messages. This can include correlationIds, or any additional information you would like
+// separate from the payload body. These attributes can be easily seen from the SQS console.
+type customAttribute struct {
+	Title string
+	// Use gosqs.DataTypeNumber or gosqs.DataTypeString
+	DataType string
+	// Value represents the value
+	Value string
+}
+
+// NewCustomAttribute adds a custom attribute to SNS and SQS messages. This can include correlationIds, logIds, or any additional information you would like
+// separate from the payload body. These attributes can be easily seen from the SQS console.
+//
+// must use gosqs.DataTypeNumber of gosqs.DataTypeString for the datatype, the value must match the type provided
+func (c *Config) NewCustomAttribute(dataType dataType, title string, value interface{}) error {
+	if dataType == DataTypeNumber {
+		val, ok := value.(int)
+		if !ok {
+			return ErrMarshal
+		}
+
+		c.Attributes = append(c.Attributes, customAttribute{title, dataType.String(), strconv.Itoa(val)})
+		return nil
+	}
+
+	val, ok := value.(string)
+	if !ok {
+		return ErrMarshal
+	}
+	c.Attributes = append(c.Attributes, customAttribute{title, dataType.String(), val})
+	return nil
+}
+
+type dataType string
+
+func (dt dataType) String() string {
+	return string(dt)
+}
+
+// DataTypeNumber represents the Number datatype, use it when creating custom attributes
+const DataTypeNumber = dataType("Number")
+
+// DataTypeString represents the String datatype, use it when creating custom attributes
+const DataTypeString = dataType("String")
+
+type retryer struct {
+	client.DefaultRetryer
+	retryCount int
+}
+
+// MaxRetries sets the total exponential back off attempts to 10 retries
+func (r retryer) MaxRetries() int {
+	if r.retryCount > 0 {
+		return r.retryCount
+	}
+
+	return 10
+}
+
+// newSession creates a new aws session
+func newSession(c Config) (*session.Session, error) {
+	//sets credentials
+	creds := credentials.NewStaticCredentials(c.Key, c.Secret, "")
+	_, err := creds.Get()
+	if err != nil {
+		return nil, ErrInvalidCreds.Context(err)
+	}
+
+	r := &retryer{retryCount: c.RetryCount}
+
+	cfg := request.WithRetryer(aws.NewConfig().WithRegion(c.Region).WithCredentials(creds), r)
+
+	//if an optional hostname config is provided, then replace the default one
+	//
+	// This will set the default AWS URL to a hostname of your choice. Perfect for testing, or mocking functionality
+	if c.Hostname != "" {
+		cfg.Endpoint = &c.Hostname
+	}
+
+	return session.NewSession(cfg)
+}