Broadway SQS

https://hexdocs.pm/broadway_sqs/v0.7.4/
Admin
Broadway SQS is an Elixir library that provides a GenStage producer for continuously polling...
Tokens:2,349
Snippets:25
Trust Score:9
Update:5 months ago
Show doc for...
Tokens:
Raw
### Start Broadway SQS Producer

Source: https://hexdocs.pm/broadway_sqs

Configure and start a Broadway SQS producer with specified queue URL, AWS credentials, and batching options. Ensure AWS credentials and region are correctly set.

```elixir
Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module: {BroadwaySQS.Producer,
      queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
      config: [
        access_key_id: "YOUR_AWS_ACCESS_KEY_ID",
        secret_access_key: "YOUR_AWS_SECRET_ACCESS_KEY",
        region: "us-east-2"
      ]
    }
  ],
  processors: [
    default: []
  ],
  batchers: [
    default: [
      batch_size: 10,
      batch_timeout: 2000
    ]
  ]
)
```

--------------------------------

### BroadwaySQS Producer Configuration

Source: https://hexdocs.pm/broadway_sqs

Example of configuring a BroadwaySQS producer with common options like queue URL, receive interval, and acknowledgment behavior.

```elixir
producer = BroadwaySQS.Producer,
  queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
  receive_interval: 3000,
  on_success: :ack,
  on_failure: :noop
```

--------------------------------

### Message Attribute Configuration

Source: https://hexdocs.pm/broadway_sqs

Example of configuring the producer to retrieve specific message attributes, such as sender ID and approximate receive count.

```elixir
producer = BroadwaySQS.Producer,
  queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
  attribute_names: [:sender_id, :approximate_receive_count]
```

--------------------------------

### Message Attribute Names Configuration

Source: https://hexdocs.pm/broadway_sqs

Example of configuring the producer to retrieve custom message attributes using a list of names or a wildcard pattern.

```elixir
producer = BroadwaySQS.Producer,
  queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
  message_attribute_names: ["custom_attribute.*", ".*"]
```

--------------------------------

### Telemetry Event: Receive Messages Start

Source: https://hexdocs.pm/broadway_sqs

This Telemetry event is dispatched before receiving messages from SQS. It includes the time of dispatch and the consumer name and demand.

```elixir
[:broadway_sqs, :receive_messages, :start]
```

--------------------------------

### Batching Configuration for Performance

Source: https://hexdocs.pm/broadway_sqs

Recommends setting a default batch size for SQS producers to improve performance and reduce costs by acknowledging messages in batches.

```elixir
batcher = Broadway.batcher(batch_size: 10)
```

--------------------------------

### Custom SQS Client Configuration

Source: https://hexdocs.pm/broadway_sqs

Demonstrates how to configure the BroadwaySQS producer to use a custom SQS client module that implements the BroadwaySQS.SQSClient behavior.

```elixir
producer = BroadwaySQS.Producer,
  queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
  sqs_client: MyCustomSQSClient
```

--------------------------------

### BroadwaySQS Batching

Source: https://hexdocs.pm/broadway_sqs

Recommendations for configuring batching in BroadwaySQS pipelines for improved performance and reduced costs.

```APIDOC
## Batching

### Description
Recommendations for batching in BroadwaySQS pipelines.

### Recommendation

It is recommended to define a default batcher with `batch_size` set to 10 for all Broadway pipelines using SQS producers. This allows messages to be acknowledged in batches, which can improve performance and reduce SQS integration costs.
```

--------------------------------

### ExAws Configuration Overrides

Source: https://hexdocs.pm/broadway_sqs

Shows how to override default ExAws configuration options for the BroadwaySQS producer, such as region and access keys.

```elixir
producer = BroadwaySQS.Producer,
  queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
  config: [
    region: "us-east-1",
    access_key_id: "YOUR_ACCESS_KEY",
    secret_access_key: "YOUR_SECRET_KEY"
  ]
```

--------------------------------

### Configure Attributes and Message Attributes

Source: https://hexdocs.pm/broadway_sqs

Configure the producer to retrieve specific SQS attributes and message attributes by setting `:attribute_names` and `:message_attribute_names`. These will be available in the message metadata.

```elixir
producer: [
  module: {BroadwaySQS.Producer,
    queue_url: "https://sqs.amazonaws.com/0000000000/my_queue",
    # Define which attributes/message_attributes you want to be attached
    attribute_names: [:approximate_receive_count],
    message_attribute_names: ["SomeAttribute"]
  }
]
```

--------------------------------

### Accessing Configured Attributes and Message Attributes

Source: https://hexdocs.pm/broadway_sqs

Access configured SQS attributes and message attributes from the message metadata in the handle_message callback. Ensure these are requested in the producer configuration.

```elixir
def handle_message(_, message, _) do
  approximate_receive_count = message.metadata.attributes["approximate_receive_count"]
  some_attribute = message.metadata.message_attributes["SomeAttribute"]

  # Do something with the attributes
end
```

--------------------------------

### BroadwaySQS.Producer Configuration

Source: https://hexdocs.pm/broadway_sqs

Configuration options for the BroadwaySQS producer. These options control how messages are received, processed, and acknowledged from an SQS queue.

```APIDOC
## BroadwaySQS.Producer Options

### Description
Configuration options for the BroadwaySQS producer.

### Options

*   `:queue_url` - Required. The url for the SQS queue. Example: `https://sqs.amazonaws.com/0000000000/my_queue`.
*   `:sqs_client` - A module that implements the `BroadwaySQS.SQSClient` behaviour. Defaults to `BroadwaySQS.ExAwsClient`.
*   `:receive_interval` (`non_neg_integer/0`) - The duration (in milliseconds) to wait before requesting more messages. Defaults to `5000`.
*   `:on_success` (`atom/0`) - Configures the acknowledgment behavior for successful messages. Possible values: `:ack`, `:noop`. Defaults to `:ack`.
*   `:on_failure` (`atom/0`) - Configures the acknowledgment behavior for failed messages. Possible values: `:ack`, `:noop`. Defaults to `:noop`.
*   `:config` (`keyword/0`) - Options to override default ExAws configuration. Commonly used: `:access_key_id`, `:secret_access_key`, `:scheme`, `:region`, `:port`. Defaults to `[]`.
*   `:max_number_of_messages` - Maximum number of messages to fetch per request (1-10). Defaults to `10`.
*   `:wait_time_seconds` - Duration (in seconds) to wait for messages (0-20). Defaults to `0`.
*   `:visibility_timeout` - Time (in seconds) a message remains invisible to other consumers (0-43200).
*   `:attribute_names` - List of attribute names to attach to message metadata. Supported values: `:sender_id`, `:sent_timestamp`, `:approximate_receive_count`, `:approximate_first_receive_timestamp`, `:sequence_number`, `:message_deduplication_id`, `:message_group_id`, `:aws_trace_header`, or `:all`.
*   `:message_attribute_names` - List of custom message attribute names to attach to message metadata. Supports wildcards `[ ".*" ]` and prefixes `[ "bar.*" ]`. Can also be `:all`.
```

--------------------------------

### Telemetry Event: Receive Messages Exception

Source: https://hexdocs.pm/broadway_sqs

This Telemetry event is dispatched after a failure while receiving messages from SQS. It includes duration, consumer name, demand, error kind, reason, and stacktrace.

```elixir
%{
  name: atom,
  demand: integer,
  kind: kind,
  reason: reason,
  stacktrace: stacktrace
}
```

--------------------------------

### Accessing Default Message Metadata

Source: https://hexdocs.pm/broadway_sqs

Access default metadata like message ID and receipt handle within the handle_message callback. This metadata is automatically attached to the %Message{} struct.

```elixir
def handle_message(_, message, _) do
  receipt = %{
    id: message.metadata.message_id,
    receipt_handle: message.metadata.receipt_handle
  }

  # Do something with the receipt
end
```

--------------------------------

### BroadwaySQS Acknowledgments

Source: https://hexdocs.pm/broadway_sqs

Controls how messages are acknowledged on SQS based on their processing outcome.

```APIDOC
## Acknowledgments

### Description
Controls the acknowledgment behavior for messages processed by BroadwaySQS.

### Options

The `:on_success` and `:on_failure` options can be used to control how messages are acknowledged.

Possible values:

*   `:ack` - Acknowledges the message, causing SQS to delete it from the queue.
*   `:noop` - Does not acknowledge the message, allowing SQS to redeliver it later based on visibility timeout and max receive count configurations.
```

--------------------------------

### Telemetry Event: Receive Messages Stop

Source: https://hexdocs.pm/broadway_sqs

This Telemetry event is dispatched after messages have been received and wrapped. It includes the duration, consumer name, messages, and demand.

```elixir
%{
  name: atom,
  messages: [Broadway.Message.t],
  demand: integer
}
```
Tokens:
Raw