consumer

Documentation

Index

• type Consumer
  • func New(queueUrl string, handler types.HandleMessage[awstypes.Message], ...) (*Consumer, func(context.Context) error, error)
  • func (c *Consumer) Start(ctx context.Context)
  • func (c *Consumer) Stop(ctx context.Context) error
• type ConsumerOptions
  • func WithAckBatchInterval(interval time.Duration) ConsumerOptions
  • func WithAckBatchSize(size int) ConsumerOptions
  • func WithConcurrency(concurrency int) ConsumerOptions
  • func WithCredentialProvider(resolver aws.CredentialsProvider) ConsumerOptions
  • func WithDispatchStrategy(strategy runner.DispatchStrategy) ConsumerOptions
  • func WithEndpointResolver(resolver sqs.EndpointResolverV2) ConsumerOptions
  • func WithHeartbeat(interval time.Duration) ConsumerOptions
  • func WithMaxIdleTime(max int) ConsumerOptions
  • func WithMaxMessages(max int) ConsumerOptions
  • func WithQueueAttributeNames(names []string) ConsumerOptions
  • func WithSqsClient(client *sqs.Client) ConsumerOptions
  • func WithVisibilityTimeout(visibility int) ConsumerOptions

Types

type Consumer

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

func New

func New(
	queueUrl string,
	handler types.HandleMessage[awstypes.Message],
	opts ...ConsumerOptions,
) (*Consumer, func(context.Context) error, error)

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context)

func (*Consumer) Stop

func (c *Consumer) Stop(ctx context.Context) error

type ConsumerOptions

type ConsumerOptions func(*Consumer)

func WithAckBatchInterval

func WithAckBatchInterval(interval time.Duration) ConsumerOptions

func WithAckBatchSize

func WithAckBatchSize(size int) ConsumerOptions

func WithConcurrency

func WithConcurrency(concurrency int) ConsumerOptions

func WithCredentialProvider

func WithCredentialProvider(resolver aws.CredentialsProvider) ConsumerOptions

Overrides how the AWS sdk gets the credientials

func WithDispatchStrategy

func WithDispatchStrategy(strategy runner.DispatchStrategy) ConsumerOptions

func WithEndpointResolver

func WithEndpointResolver(resolver sqs.EndpointResolverV2) ConsumerOptions

Overrides how the AWS sdk will resolve the SQS endpoint

func WithHeartbeat

func WithHeartbeat(interval time.Duration) ConsumerOptions

Configures the consumer to send a heartbeat to sqs for long running messages and update their visibility timeout

func WithMaxIdleTime

func WithMaxIdleTime(max int) ConsumerOptions

Sets how long (in seconds) the consumer will sit idle when there are no messages in the queue. If a messages appears in the queue while idle, it will be received immediatly

func WithMaxMessages

func WithMaxMessages(max int) ConsumerOptions

Sets the max messages to pull in at a time Default: 10

func WithQueueAttributeNames

func WithQueueAttributeNames(names []string) ConsumerOptions

Sets optional attributes to get returned from the message

func WithSqsClient

func WithSqsClient(client *sqs.Client) ConsumerOptions

Provide your own sqs client to interface with sqs

func WithVisibilityTimeout

func WithVisibilityTimeout(visibility int) ConsumerOptions

Sets the visibility timeout before SQS will requeue the message and send to other consumers Default: 60s