gosqs

type Adapter func(Handler) Handler

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
}

type Consumer interface {
	// Consume polls for new messages and if it finds one, decodes it, sends it to the handler and deletes it
	//
	// A message is not considered dequeued until it has been sucessfully processed and deleted. There is a 30 Second
	// delay between receiving a single message and receiving the same message. This delay can be adjusted in the AWS
	// console and can also be extended during operation. If a message is successfully received 4 times but not deleted,
	// it will be considered unprocessable and sent to the DLQ automatically
	//
	// Consume uses long-polling to check and retrieve messages, if it is unable to make a connection, the aws-SDK will use its
	// advanced retrying mechanism (including exponential backoff), if all of the retries fail, then we will wait 10s before
	// trying again.
	//
	// When a new message is received, it runs in a separate go-routine that will handle the full consuming of the message, error reporting
	// and deleting
	Consume()
	// RegisterHandler registers an event listener and an associated handler. If the event matches, the handler will
	// be run
	RegisterHandler(name string, h Handler, adapters ...Adapter)
	// Message serves as the direct messaging capability within the consumer. A worker can send direct messages to other workers
	Message(ctx context.Context, queue, event string, body interface{})
	// MessageSelf serves as the self messaging capability within the consumer, a worker can send messages to itself for continued
	// processing and resiliency
	MessageSelf(ctx context.Context, event string, body interface{})
}

type Handler func(context.Context, Message) error

type Logger interface {
	Println(v ...interface{})
}

type Message interface {
	// Route returns the event name that is used for routing within a worker, e.g. post_published
	Route() string
	// Decode will unmarshal the message into a supplied output using json
	Decode(out interface{}) error
	// DecodeModified is used for decoding the modification message, it will populate the body with the actual message and a
	// map[string]interface{} to view original values from that message
	DecodeModified(out interface{}, changes interface{}) error
	// Attribute will return the custom attribute that was sent through out the request.
	Attribute(key string) string
}

type Notifier interface {
	ModelName() string
}

type Publisher interface {
	// Create sends a message using a notifier, the modelname will be prepended to the static event, e.g post_created
	Create(n Notifier)
	// Delete sends a message using a notifier, the modelname will be prepended to the static event, e.g post_deleted
	Delete(n Notifier)
	// Update sends a message using a notifier, the modelname will be prepended to the static event, e.g post_updated
	Update(n Notifier)
	// Modify sends a message using a notifier, as a map of changes. The modelname will be prepended to the static event, e.g post_modified
	//
	// a special decoder will need to be used to process these events
	Modify(n Notifier, changes interface{})
	// Dispatch sends a message using a notifier, the modelname will be prepended to the provided event, e.g post_published
	Dispatch(n Notifier, event string)
	// Message sends a direct message to an individual queue, the queueName(receiver) must be provided. The event will be sent
	// as is, no prepending will take place. No other queues will receive this message.
	Message(queue, message string, body interface{})
}

type SQSError struct {
	Err string `json:"err"`
	// contains filtered or unexported fields
}