messages = receiveMessageResponse.messages();
if (messages.isEmpty()) {
throw new RuntimeException("No messages received");
}
Message message = messages.getFirst();
logger.info("\nMessage received.");
logger.info(" ID: {}", message.messageId());
logger.info(" Receipt handle: {}", message.receiptHandle());
logger.info(" Message body size: {}", message.body().length());
logger.info(" Message body (first 5 characters): {}", message.body().substring(0, 5));
return message;
} catch (RuntimeException e) {
logger.error("Error during message processing: {}", e.getMessage(), e);
throw e;
}
}
```
+ For more information, see [AWS SDK for Java 2.x Developer Guide](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-s3-messages.html).
+ For API details, see the following topics in *AWS SDK for Java 2.x API Reference*.
+ [CreateBucket](https://docs.aws.amazon.com/goto/SdkForJavaV2/s3-2006-03-01/CreateBucket)
+ [PutBucketLifecycleConfiguration](https://docs.aws.amazon.com/goto/SdkForJavaV2/s3-2006-03-01/PutBucketLifecycleConfiguration)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/ReceiveMessage)
+ [SendMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/SendMessage)
------
# Receive and process Amazon S3 event notifications by using an AWS SDK
The following code example shows how to work with S3 event notifications in an object-oriented way.
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/s3#code-examples).
This example show how to process S3 notification event by using Amazon SQS.
```
/**
* This method receives S3 event notifications by using an SqsAsyncClient.
* After the client receives the messages it deserializes the JSON payload and logs them. It uses
* the S3EventNotification class (part of the S3 event notification API for Java) to deserialize
* the JSON payload and access the messages in an object-oriented way.
*
* @param queueUrl The URL of the AWS SQS queue that receives the S3 event notifications.
* @see S3EventNotification API.
*
* To use S3 event notification serialization/deserialization to objects, add the following
* dependency to your Maven pom.xml file.
*
* software.amazon.awssdk
* s3-event-notifications
*
*
*
* The S3 event notification API became available with version 2.25.11 of the Java SDK.
*
* This example shows the use of the API with AWS SQS, but it can be used to process S3 event notifications
* in AWS SNS or AWS Lambda as well.
*
* Note: The S3EventNotification class does not work with messages routed through AWS EventBridge.
*/
static void processS3Events(String bucketName, String queueUrl, String queueArn) {
try {
// Configure the bucket to send Object Created and Object Tagging notifications to an existing SQS queue.
s3Client.putBucketNotificationConfiguration(b -> b
.notificationConfiguration(ncb -> ncb
.queueConfigurations(qcb -> qcb
.events(Event.S3_OBJECT_CREATED, Event.S3_OBJECT_TAGGING)
.queueArn(queueArn)))
.bucket(bucketName)
).join();
triggerS3EventNotifications(bucketName);
// Wait for event notifications to propagate.
Thread.sleep(Duration.ofSeconds(5).toMillis());
boolean didReceiveMessages = true;
while (didReceiveMessages) {
// Display the number of messages that are available in the queue.
sqsClient.getQueueAttributes(b -> b
.queueUrl(queueUrl)
.attributeNames(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES)
).thenAccept(attributeResponse ->
logger.info("Approximate number of messages in the queue: {}",
attributeResponse.attributes().get(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES)))
.join();
// Receive the messages.
ReceiveMessageResponse response = sqsClient.receiveMessage(b -> b
.queueUrl(queueUrl)
).get();
logger.info("Count of received messages: {}", response.messages().size());
didReceiveMessages = !response.messages().isEmpty();
// Create a collection to hold the received message for deletion
// after we log the messages.
HashSet messagesToDelete = new HashSet<>();
// Process each message.
response.messages().forEach(message -> {
logger.info("Message id: {}", message.messageId());
// Deserialize JSON message body to a S3EventNotification object
// to access messages in an object-oriented way.
S3EventNotification event = S3EventNotification.fromJson(message.body());
// Log the S3 event notification record details.
if (event.getRecords() != null) {
event.getRecords().forEach(record -> {
String eventName = record.getEventName();
String key = record.getS3().getObject().getKey();
logger.info(record.toString());
logger.info("Event name is {} and key is {}", eventName, key);
});
}
// Add logged messages to collection for batch deletion.
messagesToDelete.add(DeleteMessageBatchRequestEntry.builder()
.id(message.messageId())
.receiptHandle(message.receiptHandle())
.build());
});
// Delete messages.
if (!messagesToDelete.isEmpty()) {
sqsClient.deleteMessageBatch(DeleteMessageBatchRequest.builder()
.queueUrl(queueUrl)
.entries(messagesToDelete)
.build()
).join();
}
} // End of while block.
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException(e);
}
}
```
+ For API details, see the following topics in *AWS SDK for Java 2.x API Reference*.
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteMessageBatch)
+ [GetQueueAttributes](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/GetQueueAttributes)
+ [PutBucketNotificationConfiguration](https://docs.aws.amazon.com/goto/SdkForJavaV2/s3-2006-03-01/PutBucketNotificationConfiguration)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/ReceiveMessage)
------
# Publish Amazon SNS messages to Amazon SQS queues using an AWS SDK
The following code examples show how to:
+ Create topic (FIFO or non-FIFO).
+ Subscribe several queues to the topic with an option to apply a filter.
+ Publish messages to the topic.
+ Poll the queues for messages received.
------
#### [ .NET ]
**SDK for .NET**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/cross-service/TopicsAndQueues#code-examples).
Run an interactive scenario at a command prompt.
```
///
/// Console application to run a feature scenario for topics and queues.
///
public static class TopicsAndQueues
{
private static bool _useFifoTopic = false;
private static bool _useContentBasedDeduplication = false;
private static string _topicName = null!;
private static string _topicArn = null!;
private static readonly int _queueCount = 2;
private static readonly string[] _queueUrls = new string[_queueCount];
private static readonly string[] _subscriptionArns = new string[_queueCount];
private static readonly string[] _tones = { "cheerful", "funny", "serious", "sincere" };
public static SNSWrapper SnsWrapper { get; set; } = null!;
public static SQSWrapper SqsWrapper { get; set; } = null!;
public static bool UseConsole { get; set; } = true;
static async Task Main(string[] args)
{
// Set up dependency injection for Amazon EventBridge.
using var host = Host.CreateDefaultBuilder(args)
.ConfigureLogging(logging =>
logging.AddFilter("System", LogLevel.Debug)
.AddFilter("Microsoft", LogLevel.Information)
.AddFilter("Microsoft", LogLevel.Trace))
.ConfigureServices((_, services) =>
services.AddAWSService()
.AddAWSService()
.AddTransient()
.AddTransient()
)
.Build();
ServicesSetup(host);
PrintDescription();
await RunScenario();
}
///
/// Populate the services for use within the console application.
///
/// The services host.
private static void ServicesSetup(IHost host)
{
SnsWrapper = host.Services.GetRequiredService();
SqsWrapper = host.Services.GetRequiredService();
}
///
/// Run the scenario for working with topics and queues.
///
/// True if successful.
public static async Task RunScenario()
{
try
{
await SetupTopic();
await SetupQueues();
await PublishMessages();
foreach (var queueUrl in _queueUrls)
{
var messages = await PollForMessages(queueUrl);
if (messages.Any())
{
await DeleteMessages(queueUrl, messages);
}
}
await CleanupResources();
Console.WriteLine("Messaging with topics and queues scenario is complete.");
return true;
}
catch (Exception ex)
{
Console.WriteLine(new string('-', 80));
Console.WriteLine($"There was a problem running the scenario: {ex.Message}");
await CleanupResources();
Console.WriteLine(new string('-', 80));
return false;
}
}
///
/// Print a description for the tasks in the scenario.
///
/// Async task.
private static void PrintDescription()
{
Console.WriteLine(new string('-', 80));
Console.WriteLine($"Welcome to messaging with topics and queues.");
Console.WriteLine(new string('-', 80));
Console.WriteLine($"In this scenario, you will create an SNS topic and subscribe {_queueCount} SQS queues to the topic." +
$"\r\nYou can select from several options for configuring the topic and the subscriptions for the 2 queues." +
$"\r\nYou can then post to the topic and see the results in the queues.\r\n");
Console.WriteLine(new string('-', 80));
}
///
/// Set up the SNS topic to be used with the queues.
///
/// Async task.
private static async Task SetupTopic()
{
Console.WriteLine(new string('-', 80));
Console.WriteLine($"SNS topics can be configured as FIFO (First-In-First-Out)." +
$"\r\nFIFO topics deliver messages in order and support deduplication and message filtering." +
$"\r\nYou can then post to the topic and see the results in the queues.\r\n");
_useFifoTopic = GetYesNoResponse("Would you like to work with FIFO topics?");
if (_useFifoTopic)
{
Console.WriteLine(new string('-', 80));
_topicName = GetUserResponse("Enter a name for your SNS topic: ", "example-topic");
Console.WriteLine(
"Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.\r\n");
Console.WriteLine(new string('-', 80));
Console.WriteLine($"Because you have chosen a FIFO topic, deduplication is supported." +
$"\r\nDeduplication IDs are either set in the message or automatically generated " +
$"\r\nfrom content using a hash function.\r\n" +
$"\r\nIf a message is successfully published to an SNS FIFO topic, any message " +
$"\r\npublished and determined to have the same deduplication ID, " +
$"\r\nwithin the five-minute deduplication interval, is accepted but not delivered.\r\n" +
$"\r\nFor more information about deduplication, " +
$"\r\nsee https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html.");
_useContentBasedDeduplication = GetYesNoResponse("Use content-based deduplication instead of entering a deduplication ID?");
Console.WriteLine(new string('-', 80));
}
_topicArn = await SnsWrapper.CreateTopicWithName(_topicName, _useFifoTopic, _useContentBasedDeduplication);
Console.WriteLine($"Your new topic with the name {_topicName}" +
$"\r\nand Amazon Resource Name (ARN) {_topicArn}" +
$"\r\nhas been created.\r\n");
Console.WriteLine(new string('-', 80));
return _topicArn;
}
///
/// Set up the queues.
///
/// Async task.
private static async Task SetupQueues()
{
Console.WriteLine(new string('-', 80));
Console.WriteLine($"Now you will create {_queueCount} Amazon Simple Queue Service (Amazon SQS) queues to subscribe to the topic.");
// Repeat this section for each queue.
for (int i = 0; i < _queueCount; i++)
{
var queueName = GetUserResponse("Enter a name for an Amazon SQS queue: ", $"example-queue-{i}");
if (_useFifoTopic)
{
// Only explain this once.
if (i == 0)
{
Console.WriteLine(
"Because you have selected a FIFO topic, '.fifo' must be appended to the queue name.");
}
var queueUrl = await SqsWrapper.CreateQueueWithName(queueName, _useFifoTopic);
_queueUrls[i] = queueUrl;
Console.WriteLine($"Your new queue with the name {queueName}" +
$"\r\nand queue URL {queueUrl}" +
$"\r\nhas been created.\r\n");
if (i == 0)
{
Console.WriteLine(
$"The queue URL is used to retrieve the queue ARN,\r\n" +
$"which is used to create a subscription.");
Console.WriteLine(new string('-', 80));
}
var queueArn = await SqsWrapper.GetQueueArnByUrl(queueUrl);
if (i == 0)
{
Console.WriteLine(
$"An AWS Identity and Access Management (IAM) policy must be attached to an SQS queue, enabling it to receive\r\n" +
$"messages from an SNS topic");
}
await SqsWrapper.SetQueuePolicyForTopic(queueArn, _topicArn, queueUrl);
await SetupFilters(i, queueArn, queueName);
}
}
Console.WriteLine(new string('-', 80));
}
///
/// Set up filters with user options for a queue.
///
/// The number of this queue.
/// The ARN of the queue.
/// The name of the queue.
/// Async Task.
public static async Task SetupFilters(int queueCount, string queueArn, string queueName)
{
if (_useFifoTopic)
{
Console.WriteLine(new string('-', 80));
// Only explain this once.
if (queueCount == 0)
{
Console.WriteLine(
"Subscriptions to a FIFO topic can have filters." +
"If you add a filter to this subscription, then only the filtered messages " +
"will be received in the queue.");
Console.WriteLine(
"For information about message filtering, " +
"see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html");
Console.WriteLine(
"For this example, you can filter messages by a" +
"TONE attribute.");
}
var useFilter = GetYesNoResponse($"Filter messages for {queueName}'s subscription to the topic?");
string? filterPolicy = null;
if (useFilter)
{
filterPolicy = CreateFilterPolicy();
}
var subscriptionArn = await SnsWrapper.SubscribeTopicWithFilter(_topicArn, filterPolicy,
queueArn);
_subscriptionArns[queueCount] = subscriptionArn;
Console.WriteLine(
$"The queue {queueName} has been subscribed to the topic {_topicName} " +
$"with the subscription ARN {subscriptionArn}");
Console.WriteLine(new string('-', 80));
}
}
///
/// Use user input to create a filter policy for a subscription.
///
/// The serialized filter policy.
public static string CreateFilterPolicy()
{
Console.WriteLine(new string('-', 80));
Console.WriteLine(
$"You can filter messages by one or more of the following" +
$"TONE attributes.");
List filterSelections = new List();
var selectionNumber = 0;
do
{
Console.WriteLine(
$"Enter a number to add a TONE filter, or enter 0 to stop adding filters.");
for (int i = 0; i < _tones.Length; i++)
{
Console.WriteLine($"\t{i + 1}. {_tones[i]}");
}
var selection = GetUserResponse("", filterSelections.Any() ? "0" : "1");
int.TryParse(selection, out selectionNumber);
if (selectionNumber > 0 && !filterSelections.Contains(_tones[selectionNumber - 1]))
{
filterSelections.Add(_tones[selectionNumber - 1]);
}
} while (selectionNumber != 0);
var filters = new Dictionary>
{
{ "tone", filterSelections }
};
string filterPolicy = JsonSerializer.Serialize(filters);
return filterPolicy;
}
///
/// Publish messages using user settings.
///
/// Async task.
public static async Task PublishMessages()
{
Console.WriteLine("Now we can publish messages.");
var keepSendingMessages = true;
string? deduplicationId = null;
string? toneAttribute = null;
while (keepSendingMessages)
{
Console.WriteLine();
var message = GetUserResponse("Enter a message to publish.", "This is a sample message");
if (_useFifoTopic)
{
Console.WriteLine("Because you are using a FIFO topic, you must set a message group ID." +
"\r\nAll messages within the same group will be received in the order " +
"they were published.");
Console.WriteLine();
var messageGroupId = GetUserResponse("Enter a message group ID for this message:", "1");
if (!_useContentBasedDeduplication)
{
Console.WriteLine("Because you are not using content-based deduplication, " +
"you must enter a deduplication ID.");
Console.WriteLine("Enter a deduplication ID for this message.");
deduplicationId = GetUserResponse("Enter a deduplication ID for this message.", "1");
}
if (GetYesNoResponse("Add an attribute to this message?"))
{
Console.WriteLine("Enter a number for an attribute.");
for (int i = 0; i < _tones.Length; i++)
{
Console.WriteLine($"\t{i + 1}. {_tones[i]}");
}
var selection = GetUserResponse("", "1");
int.TryParse(selection, out var selectionNumber);
if (selectionNumber > 0 && selectionNumber < _tones.Length)
{
toneAttribute = _tones[selectionNumber - 1];
}
}
var messageID = await SnsWrapper.PublishToTopicWithAttribute(
_topicArn, message, "tone", toneAttribute, deduplicationId, messageGroupId);
Console.WriteLine($"Message published with id {messageID}.");
}
keepSendingMessages = GetYesNoResponse("Send another message?", false);
}
}
///
/// Poll for the published messages to see the results of the user's choices.
///
/// Async task.
public static async Task> PollForMessages(string queueUrl)
{
Console.WriteLine(new string('-', 80));
Console.WriteLine($"Now the SQS queue at {queueUrl} will be polled to retrieve the messages." +
"\r\nPress any key to continue.");
if (UseConsole)
{
Console.ReadLine();
}
var moreMessages = true;
var messages = new List();
while (moreMessages)
{
var newMessages = await SqsWrapper.ReceiveMessagesByUrl(queueUrl, 10);
moreMessages = newMessages.Any();
if (moreMessages)
{
messages.AddRange(newMessages);
}
}
Console.WriteLine($"{messages.Count} message(s) were received by the queue at {queueUrl}.");
foreach (var message in messages)
{
Console.WriteLine("\tMessage:" +
$"\n\t{message.Body}");
}
Console.WriteLine(new string('-', 80));
return messages;
}
///
/// Delete the message using handles in a batch.
///
/// Async task.
public static async Task DeleteMessages(string queueUrl, List messages)
{
Console.WriteLine(new string('-', 80));
Console.WriteLine("Now we can delete the messages in this queue in a batch.");
await SqsWrapper.DeleteMessageBatchByUrl(queueUrl, messages);
Console.WriteLine(new string('-', 80));
}
///
/// Clean up the resources from the scenario.
///
/// Async task.
private static async Task CleanupResources()
{
Console.WriteLine(new string('-', 80));
Console.WriteLine($"Clean up resources.");
try
{
foreach (var queueUrl in _queueUrls)
{
if (!string.IsNullOrEmpty(queueUrl))
{
var deleteQueue =
GetYesNoResponse($"Delete queue with url {queueUrl}?");
if (deleteQueue)
{
await SqsWrapper.DeleteQueueByUrl(queueUrl);
}
}
}
foreach (var subscriptionArn in _subscriptionArns)
{
if (!string.IsNullOrEmpty(subscriptionArn))
{
await SnsWrapper.UnsubscribeByArn(subscriptionArn);
}
}
var deleteTopic = GetYesNoResponse($"Delete topic {_topicName}?");
if (deleteTopic)
{
await SnsWrapper.DeleteTopicByArn(_topicArn);
}
}
catch (Exception ex)
{
Console.WriteLine($"Unable to clean up resources. Here's why: {ex.Message}.");
}
Console.WriteLine(new string('-', 80));
}
///
/// Helper method to get a yes or no response from the user.
///
/// The question string to print on the console.
/// Optional default answer to use.
/// True if the user responds with a yes.
private static bool GetYesNoResponse(string question, bool defaultAnswer = true)
{
if (UseConsole)
{
Console.WriteLine(question);
var ynResponse = Console.ReadLine();
var response = ynResponse != null &&
ynResponse.Equals("y",
StringComparison.InvariantCultureIgnoreCase);
return response;
}
// If not using the console, use the default.
return defaultAnswer;
}
///
/// Helper method to get a string response from the user through the console.
///
/// The question string to print on the console.
/// Optional default answer to use.
/// True if the user responds with a yes.
private static string GetUserResponse(string question, string defaultAnswer)
{
if (UseConsole)
{
var response = "";
while (string.IsNullOrEmpty(response))
{
Console.WriteLine(question);
response = Console.ReadLine();
}
return response;
}
// If not using the console, use the default.
return defaultAnswer;
}
}
```
Create a class that wraps Amazon SQS operations.
```
///
/// Wrapper for Amazon Simple Queue Service (SQS) operations.
///
public class SQSWrapper
{
private readonly IAmazonSQS _amazonSQSClient;
///
/// Constructor for the Amazon SQS wrapper.
///
/// The injected Amazon SQS client.
public SQSWrapper(IAmazonSQS amazonSQS)
{
_amazonSQSClient = amazonSQS;
}
///
/// Create a queue with a specific name.
///
/// The name for the queue.
/// True to use a FIFO queue.
/// The url for the queue.
public async Task CreateQueueWithName(string queueName, bool useFifoQueue)
{
int maxMessage = 256 * 1024;
var queueAttributes = new Dictionary
{
{
QueueAttributeName.MaximumMessageSize,
maxMessage.ToString()
}
};
var createQueueRequest = new CreateQueueRequest()
{
QueueName = queueName,
Attributes = queueAttributes
};
if (useFifoQueue)
{
// Update the name if it is not correct for a FIFO queue.
if (!queueName.EndsWith(".fifo"))
{
createQueueRequest.QueueName = queueName + ".fifo";
}
// Add an attribute for a FIFO queue.
createQueueRequest.Attributes.Add(
QueueAttributeName.FifoQueue, "true");
}
var createResponse = await _amazonSQSClient.CreateQueueAsync(
new CreateQueueRequest()
{
QueueName = queueName
});
return createResponse.QueueUrl;
}
///
/// Get the ARN for a queue from its URL.
///
/// The URL of the queue.
/// The ARN of the queue.
public async Task GetQueueArnByUrl(string queueUrl)
{
var getAttributesRequest = new GetQueueAttributesRequest()
{
QueueUrl = queueUrl,
AttributeNames = new List() { QueueAttributeName.QueueArn }
};
var getAttributesResponse = await _amazonSQSClient.GetQueueAttributesAsync(
getAttributesRequest);
return getAttributesResponse.QueueARN;
}
///
/// Set the policy attribute of a queue for a topic.
///
/// The ARN of the queue.
/// The ARN of the topic.
/// The url for the queue.
/// True if successful.
public async Task SetQueuePolicyForTopic(string queueArn, string topicArn, string queueUrl)
{
var queuePolicy = "{" +
"\"Version\": \"2012-10-17\"," +
"\"Statement\": [{" +
"\"Effect\": \"Allow\"," +
"\"Principal\": {" +
$"\"Service\": " +
"\"sns.amazonaws.com\"" +
"}," +
"\"Action\": \"sqs:SendMessage\"," +
$"\"Resource\": \"{queueArn}\"," +
"\"Condition\": {" +
"\"ArnEquals\": {" +
$"\"aws:SourceArn\": \"{topicArn}\"" +
"}" +
"}" +
"}]" +
"}";
var attributesResponse = await _amazonSQSClient.SetQueueAttributesAsync(
new SetQueueAttributesRequest()
{
QueueUrl = queueUrl,
Attributes = new Dictionary() { { "Policy", queuePolicy } }
});
return attributesResponse.HttpStatusCode == HttpStatusCode.OK;
}
///
/// Receive messages from a queue by its URL.
///
/// The url of the queue.
/// The list of messages.
public async Task> ReceiveMessagesByUrl(string queueUrl, int maxMessages)
{
// Setting WaitTimeSeconds to non-zero enables long polling.
// For information about long polling, see
// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html
var messageResponse = await _amazonSQSClient.ReceiveMessageAsync(
new ReceiveMessageRequest()
{
QueueUrl = queueUrl,
MaxNumberOfMessages = maxMessages,
WaitTimeSeconds = 1
});
return messageResponse.Messages;
}
///
/// Delete a batch of messages from a queue by its url.
///
/// The url of the queue.
/// True if successful.
public async Task DeleteMessageBatchByUrl(string queueUrl, List messages)
{
var deleteRequest = new DeleteMessageBatchRequest()
{
QueueUrl = queueUrl,
Entries = new List()
};
foreach (var message in messages)
{
deleteRequest.Entries.Add(new DeleteMessageBatchRequestEntry()
{
ReceiptHandle = message.ReceiptHandle,
Id = message.MessageId
});
}
var deleteResponse = await _amazonSQSClient.DeleteMessageBatchAsync(deleteRequest);
return deleteResponse.Failed.Any();
}
///
/// Delete a queue by its URL.
///
/// The url of the queue.
/// True if successful.
public async Task DeleteQueueByUrl(string queueUrl)
{
var deleteResponse = await _amazonSQSClient.DeleteQueueAsync(
new DeleteQueueRequest()
{
QueueUrl = queueUrl
});
return deleteResponse.HttpStatusCode == HttpStatusCode.OK;
}
}
```
Create a class that wraps Amazon SNS operations.
```
///
/// Wrapper for Amazon Simple Notification Service (SNS) operations.
///
public class SNSWrapper
{
private readonly IAmazonSimpleNotificationService _amazonSNSClient;
///
/// Constructor for the Amazon SNS wrapper.
///
/// The injected Amazon SNS client.
public SNSWrapper(IAmazonSimpleNotificationService amazonSNS)
{
_amazonSNSClient = amazonSNS;
}
///
/// Create a new topic with a name and specific FIFO and de-duplication attributes.
///
/// The name for the topic.
/// True to use a FIFO topic.
/// True to use content-based de-duplication.
/// The ARN of the new topic.
public async Task CreateTopicWithName(string topicName, bool useFifoTopic, bool useContentBasedDeduplication)
{
var createTopicRequest = new CreateTopicRequest()
{
Name = topicName,
};
if (useFifoTopic)
{
// Update the name if it is not correct for a FIFO topic.
if (!topicName.EndsWith(".fifo"))
{
createTopicRequest.Name = topicName + ".fifo";
}
// Add the attributes from the method parameters.
createTopicRequest.Attributes = new Dictionary
{
{ "FifoTopic", "true" }
};
if (useContentBasedDeduplication)
{
createTopicRequest.Attributes.Add("ContentBasedDeduplication", "true");
}
}
var createResponse = await _amazonSNSClient.CreateTopicAsync(createTopicRequest);
return createResponse.TopicArn;
}
///
/// Subscribe a queue to a topic with optional filters.
///
/// The ARN of the topic.
/// The optional filtering policy for the subscription.
/// The ARN of the queue.
/// The ARN of the new subscription.
public async Task SubscribeTopicWithFilter(string topicArn, string? filterPolicy, string queueArn)
{
var subscribeRequest = new SubscribeRequest()
{
TopicArn = topicArn,
Protocol = "sqs",
Endpoint = queueArn
};
if (!string.IsNullOrEmpty(filterPolicy))
{
subscribeRequest.Attributes = new Dictionary { { "FilterPolicy", filterPolicy } };
}
var subscribeResponse = await _amazonSNSClient.SubscribeAsync(subscribeRequest);
return subscribeResponse.SubscriptionArn;
}
///
/// Publish a message to a topic with an attribute and optional deduplication and group IDs.
///
/// The ARN of the topic.
/// The message to publish.
/// The optional attribute for the message.
/// The optional attribute value for the message.
/// The optional deduplication ID for the message.
/// The optional group ID for the message.
/// The ID of the message published.
public async Task PublishToTopicWithAttribute(
string topicArn,
string message,
string? attributeName = null,
string? attributeValue = null,
string? deduplicationId = null,
string? groupId = null)
{
var publishRequest = new PublishRequest()
{
TopicArn = topicArn,
Message = message,
MessageDeduplicationId = deduplicationId,
MessageGroupId = groupId
};
if (attributeValue != null)
{
// Add the string attribute if it exists.
publishRequest.MessageAttributes =
new Dictionary
{
{ attributeName!, new MessageAttributeValue() { StringValue = attributeValue, DataType = "String"} }
};
}
var publishResponse = await _amazonSNSClient.PublishAsync(publishRequest);
return publishResponse.MessageId;
}
///
/// Unsubscribe from a topic by a subscription ARN.
///
/// The ARN of the subscription.
/// True if successful.
public async Task UnsubscribeByArn(string subscriptionArn)
{
var unsubscribeResponse = await _amazonSNSClient.UnsubscribeAsync(
new UnsubscribeRequest()
{
SubscriptionArn = subscriptionArn
});
return unsubscribeResponse.HttpStatusCode == HttpStatusCode.OK;
}
///
/// Delete a topic by its topic ARN.
///
/// The ARN of the topic.
/// True if successful.
public async Task DeleteTopicByArn(string topicArn)
{
var deleteResponse = await _amazonSNSClient.DeleteTopicAsync(
new DeleteTopicRequest()
{
TopicArn = topicArn
});
return deleteResponse.HttpStatusCode == HttpStatusCode.OK;
}
}
```
+ For API details, see the following topics in *AWS SDK for .NET API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/DotNetSDKV3/sqs-2012-11-05/CreateQueue)
+ [CreateTopic](https://docs.aws.amazon.com/goto/DotNetSDKV3/sns-2010-03-31/CreateTopic)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/DotNetSDKV3/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/DotNetSDKV3/sqs-2012-11-05/DeleteQueue)
+ [DeleteTopic](https://docs.aws.amazon.com/goto/DotNetSDKV3/sns-2010-03-31/DeleteTopic)
+ [GetQueueAttributes](https://docs.aws.amazon.com/goto/DotNetSDKV3/sqs-2012-11-05/GetQueueAttributes)
+ [Publish](https://docs.aws.amazon.com/goto/DotNetSDKV3/sns-2010-03-31/Publish)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/DotNetSDKV3/sqs-2012-11-05/ReceiveMessage)
+ [SetQueueAttributes](https://docs.aws.amazon.com/goto/DotNetSDKV3/sqs-2012-11-05/SetQueueAttributes)
+ [Subscribe](https://docs.aws.amazon.com/goto/DotNetSDKV3/sns-2010-03-31/Subscribe)
+ [Unsubscribe](https://docs.aws.amazon.com/goto/DotNetSDKV3/sns-2010-03-31/Unsubscribe)
------
#### [ C\$1\$1 ]
**SDK for C\$1\$1**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/cpp/example_code/cross-service/topics_and_queues#code-examples).
```
Aws::Client::ClientConfiguration clientConfig;
// Optional: Set to the AWS Region (overrides config file).
// clientConfig.region = "us-east-1";
//! Workflow for messaging with topics and queues using Amazon SNS and Amazon SQS.
/*!
\param clientConfig Aws client configuration.
\return bool: Successful completion.
*/
bool AwsDoc::TopicsAndQueues::messagingWithTopicsAndQueues(
const Aws::Client::ClientConfiguration &clientConfiguration) {
std::cout << "Welcome to messaging with topics and queues." << std::endl;
printAsterisksLine();
std::cout << "In this workflow, you will create an SNS topic and subscribe "
<< NUMBER_OF_QUEUES <<
" SQS queues to the topic." << std::endl;
std::cout
<< "You can select from several options for configuring the topic and the subscriptions for the "
<< NUMBER_OF_QUEUES << " queues." << std::endl;
std::cout << "You can then post to the topic and see the results in the queues."
<< std::endl;
Aws::SNS::SNSClient snsClient(clientConfiguration);
printAsterisksLine();
std::cout << "SNS topics can be configured as FIFO (First-In-First-Out)."
<< std::endl;
std::cout
<< "FIFO topics deliver messages in order and support deduplication and message filtering."
<< std::endl;
bool isFifoTopic = askYesNoQuestion(
"Would you like to work with FIFO topics? (y/n) ");
bool contentBasedDeduplication = false;
Aws::String topicName;
if (isFifoTopic) {
printAsterisksLine();
std::cout << "Because you have chosen a FIFO topic, deduplication is supported."
<< std::endl;
std::cout
<< "Deduplication IDs are either set in the message or automatically generated "
<< "from content using a hash function." << std::endl;
std::cout
<< "If a message is successfully published to an SNS FIFO topic, any message "
<< "published and determined to have the same deduplication ID, "
<< std::endl;
std::cout
<< "within the five-minute deduplication interval, is accepted but not delivered."
<< std::endl;
std::cout
<< "For more information about deduplication, "
<< "see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html."
<< std::endl;
contentBasedDeduplication = askYesNoQuestion(
"Use content-based deduplication instead of entering a deduplication ID? (y/n) ");
}
printAsterisksLine();
Aws::SQS::SQSClient sqsClient(clientConfiguration);
Aws::Vector queueURLS;
Aws::Vector subscriptionARNS;
Aws::String topicARN;
{
topicName = askQuestion("Enter a name for your SNS topic. ");
// 1. Create an Amazon SNS topic, either FIFO or non-FIFO.
Aws::SNS::Model::CreateTopicRequest request;
if (isFifoTopic) {
request.AddAttributes("FifoTopic", "true");
if (contentBasedDeduplication) {
request.AddAttributes("ContentBasedDeduplication", "true");
}
topicName = topicName + FIFO_SUFFIX;
std::cout
<< "Because you have selected a FIFO topic, '.fifo' must be appended to the topic name."
<< std::endl;
}
request.SetName(topicName);
Aws::SNS::Model::CreateTopicOutcome outcome = snsClient.CreateTopic(request);
if (outcome.IsSuccess()) {
topicARN = outcome.GetResult().GetTopicArn();
std::cout << "Your new topic with the name '" << topicName
<< "' and the topic Amazon Resource Name (ARN) " << std::endl;
std::cout << "'" << topicARN << "' has been created." << std::endl;
}
else {
std::cerr << "Error with TopicsAndQueues::CreateTopic. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
printAsterisksLine();
std::cout << "Now you will create " << NUMBER_OF_QUEUES
<< " SQS queues to subscribe to the topic." << std::endl;
Aws::Vector queueNames;
bool filteringMessages = false;
bool first = true;
for (int i = 1; i <= NUMBER_OF_QUEUES; ++i) {
Aws::String queueURL;
Aws::String queueName;
{
printAsterisksLine();
std::ostringstream ostringstream;
ostringstream << "Enter a name for " << (first ? "an" : "the next")
<< " SQS queue. ";
queueName = askQuestion(ostringstream.str());
// 2. Create an SQS queue.
Aws::SQS::Model::CreateQueueRequest request;
if (isFifoTopic) {
request.AddAttributes(Aws::SQS::Model::QueueAttributeName::FifoQueue,
"true");
queueName = queueName + FIFO_SUFFIX;
if (first) // Only explain this once.
{
std::cout
<< "Because you are creating a FIFO SQS queue, '.fifo' must "
<< "be appended to the queue name." << std::endl;
}
}
request.SetQueueName(queueName);
queueNames.push_back(queueName);
Aws::SQS::Model::CreateQueueOutcome outcome =
sqsClient.CreateQueue(request);
if (outcome.IsSuccess()) {
queueURL = outcome.GetResult().GetQueueUrl();
std::cout << "Your new SQS queue with the name '" << queueName
<< "' and the queue URL " << std::endl;
std::cout << "'" << queueURL << "' has been created." << std::endl;
}
else {
std::cerr << "Error with SQS::CreateQueue. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
queueURLS.push_back(queueURL);
if (first) // Only explain this once.
{
std::cout
<< "The queue URL is used to retrieve the queue ARN, which is "
<< "used to create a subscription." << std::endl;
}
Aws::String queueARN;
{
// 3. Get the SQS queue ARN attribute.
Aws::SQS::Model::GetQueueAttributesRequest request;
request.SetQueueUrl(queueURL);
request.AddAttributeNames(Aws::SQS::Model::QueueAttributeName::QueueArn);
Aws::SQS::Model::GetQueueAttributesOutcome outcome =
sqsClient.GetQueueAttributes(request);
if (outcome.IsSuccess()) {
const Aws::Map &attributes =
outcome.GetResult().GetAttributes();
const auto &iter = attributes.find(
Aws::SQS::Model::QueueAttributeName::QueueArn);
if (iter != attributes.end()) {
queueARN = iter->second;
std::cout << "The queue ARN '" << queueARN
<< "' has been retrieved."
<< std::endl;
}
else {
std::cerr
<< "Error ARN attribute not returned by GetQueueAttribute."
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
else {
std::cerr << "Error with SQS::GetQueueAttributes. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
if (first) {
std::cout
<< "An IAM policy must be attached to an SQS queue, enabling it to receive "
"messages from an SNS topic." << std::endl;
}
{
// 4. Set the SQS queue policy attribute with a policy enabling the receipt of SNS messages.
Aws::SQS::Model::SetQueueAttributesRequest request;
request.SetQueueUrl(queueURL);
Aws::String policy = createPolicyForQueue(queueARN, topicARN);
request.AddAttributes(Aws::SQS::Model::QueueAttributeName::Policy,
policy);
Aws::SQS::Model::SetQueueAttributesOutcome outcome =
sqsClient.SetQueueAttributes(request);
if (outcome.IsSuccess()) {
std::cout << "The attributes for the queue '" << queueName
<< "' were successfully updated." << std::endl;
}
else {
std::cerr << "Error with SQS::SetQueueAttributes. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
printAsterisksLine();
{
// 5. Subscribe the SQS queue to the SNS topic.
Aws::SNS::Model::SubscribeRequest request;
request.SetTopicArn(topicARN);
request.SetProtocol("sqs");
request.SetEndpoint(queueARN);
if (isFifoTopic) {
if (first) {
std::cout << "Subscriptions to a FIFO topic can have filters."
<< std::endl;
std::cout
<< "If you add a filter to this subscription, then only the filtered messages "
<< "will be received in the queue." << std::endl;
std::cout << "For information about message filtering, "
<< "see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html"
<< std::endl;
std::cout << "For this example, you can filter messages by a \""
<< TONE_ATTRIBUTE << "\" attribute." << std::endl;
}
std::ostringstream ostringstream;
ostringstream << "Filter messages for \"" << queueName
<< "\"'s subscription to the topic \""
<< topicName << "\"? (y/n)";
// Add filter if user answers yes.
if (askYesNoQuestion(ostringstream.str())) {
Aws::String jsonPolicy = getFilterPolicyFromUser();
if (!jsonPolicy.empty()) {
filteringMessages = true;
std::cout << "This is the filter policy for this subscription."
<< std::endl;
std::cout << jsonPolicy << std::endl;
request.AddAttributes("FilterPolicy", jsonPolicy);
}
else {
std::cout
<< "Because you did not select any attributes, no filter "
<< "will be added to this subscription." << std::endl;
}
}
} // if (isFifoTopic)
Aws::SNS::Model::SubscribeOutcome outcome = snsClient.Subscribe(request);
if (outcome.IsSuccess()) {
Aws::String subscriptionARN = outcome.GetResult().GetSubscriptionArn();
std::cout << "The queue '" << queueName
<< "' has been subscribed to the topic '"
<< "'" << topicName << "'" << std::endl;
std::cout << "with the subscription ARN '" << subscriptionARN << "."
<< std::endl;
subscriptionARNS.push_back(subscriptionARN);
}
else {
std::cerr << "Error with TopicsAndQueues::Subscribe. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
first = false;
}
first = true;
do {
printAsterisksLine();
// 6. Publish a message to the SNS topic.
Aws::SNS::Model::PublishRequest request;
request.SetTopicArn(topicARN);
Aws::String message = askQuestion("Enter a message text to publish. ");
request.SetMessage(message);
if (isFifoTopic) {
if (first) {
std::cout
<< "Because you are using a FIFO topic, you must set a message group ID."
<< std::endl;
std::cout
<< "All messages within the same group will be received in the "
<< "order they were published." << std::endl;
}
Aws::String messageGroupID = askQuestion(
"Enter a message group ID for this message. ");
request.SetMessageGroupId(messageGroupID);
if (!contentBasedDeduplication) {
if (first) {
std::cout
<< "Because you are not using content-based deduplication, "
<< "you must enter a deduplication ID." << std::endl;
}
Aws::String deduplicationID = askQuestion(
"Enter a deduplication ID for this message. ");
request.SetMessageDeduplicationId(deduplicationID);
}
}
if (filteringMessages && askYesNoQuestion(
"Add an attribute to this message? (y/n) ")) {
for (size_t i = 0; i < TONES.size(); ++i) {
std::cout << " " << (i + 1) << ". " << TONES[i] << std::endl;
}
int selection = askQuestionForIntRange(
"Enter a number for an attribute. ",
1, static_cast(TONES.size()));
Aws::SNS::Model::MessageAttributeValue messageAttributeValue;
messageAttributeValue.SetDataType("String");
messageAttributeValue.SetStringValue(TONES[selection - 1]);
request.AddMessageAttributes(TONE_ATTRIBUTE, messageAttributeValue);
}
Aws::SNS::Model::PublishOutcome outcome = snsClient.Publish(request);
if (outcome.IsSuccess()) {
std::cout << "Your message was successfully published." << std::endl;
}
else {
std::cerr << "Error with TopicsAndQueues::Publish. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
first = false;
} while (askYesNoQuestion("Post another message? (y/n) "));
printAsterisksLine();
std::cout << "Now the SQS queue will be polled to retrieve the messages."
<< std::endl;
askQuestion("Press any key to continue...", alwaysTrueTest);
for (size_t i = 0; i < queueURLS.size(); ++i) {
// 7. Poll an SQS queue for its messages.
std::vector messages;
std::vector receiptHandles;
while (true) {
Aws::SQS::Model::ReceiveMessageRequest request;
request.SetMaxNumberOfMessages(10);
request.SetQueueUrl(queueURLS[i]);
// Setting WaitTimeSeconds to non-zero enables long polling.
// For information about long polling, see
// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html
request.SetWaitTimeSeconds(1);
Aws::SQS::Model::ReceiveMessageOutcome outcome =
sqsClient.ReceiveMessage(request);
if (outcome.IsSuccess()) {
const Aws::Vector &newMessages = outcome.GetResult().GetMessages();
if (newMessages.empty()) {
break;
}
else {
for (const Aws::SQS::Model::Message &message: newMessages) {
messages.push_back(message.GetBody());
receiptHandles.push_back(message.GetReceiptHandle());
}
}
}
else {
std::cerr << "Error with SQS::ReceiveMessage. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
printAsterisksLine();
if (messages.empty()) {
std::cout << "No messages were ";
}
else if (messages.size() == 1) {
std::cout << "One message was ";
}
else {
std::cout << messages.size() << " messages were ";
}
std::cout << "received by the queue '" << queueNames[i]
<< "'." << std::endl;
for (const Aws::String &message: messages) {
std::cout << " Message : '" << message << "'."
<< std::endl;
}
// 8. Delete a batch of messages from an SQS queue.
if (!receiptHandles.empty()) {
Aws::SQS::Model::DeleteMessageBatchRequest request;
request.SetQueueUrl(queueURLS[i]);
int id = 1; // Ids must be unique within a batch delete request.
for (const Aws::String &receiptHandle: receiptHandles) {
Aws::SQS::Model::DeleteMessageBatchRequestEntry entry;
entry.SetId(std::to_string(id));
++id;
entry.SetReceiptHandle(receiptHandle);
request.AddEntries(entry);
}
Aws::SQS::Model::DeleteMessageBatchOutcome outcome =
sqsClient.DeleteMessageBatch(request);
if (outcome.IsSuccess()) {
std::cout << "The batch deletion of messages was successful."
<< std::endl;
}
else {
std::cerr << "Error with SQS::DeleteMessageBatch. "
<< outcome.GetError().GetMessage()
<< std::endl;
cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient);
return false;
}
}
}
return cleanUp(topicARN,
queueURLS,
subscriptionARNS,
snsClient,
sqsClient,
true); // askUser
}
bool AwsDoc::TopicsAndQueues::cleanUp(const Aws::String &topicARN,
const Aws::Vector &queueURLS,
const Aws::Vector &subscriptionARNS,
const Aws::SNS::SNSClient &snsClient,
const Aws::SQS::SQSClient &sqsClient,
bool askUser) {
bool result = true;
printAsterisksLine();
if (!queueURLS.empty() && askUser &&
askYesNoQuestion("Delete the SQS queues? (y/n) ")) {
for (const auto &queueURL: queueURLS) {
// 9. Delete an SQS queue.
Aws::SQS::Model::DeleteQueueRequest request;
request.SetQueueUrl(queueURL);
Aws::SQS::Model::DeleteQueueOutcome outcome =
sqsClient.DeleteQueue(request);
if (outcome.IsSuccess()) {
std::cout << "The queue with URL '" << queueURL
<< "' was successfully deleted." << std::endl;
}
else {
std::cerr << "Error with SQS::DeleteQueue. "
<< outcome.GetError().GetMessage()
<< std::endl;
result = false;
}
}
for (const auto &subscriptionARN: subscriptionARNS) {
// 10. Unsubscribe an SNS subscription.
Aws::SNS::Model::UnsubscribeRequest request;
request.SetSubscriptionArn(subscriptionARN);
Aws::SNS::Model::UnsubscribeOutcome outcome =
snsClient.Unsubscribe(request);
if (outcome.IsSuccess()) {
std::cout << "Unsubscribe of subscription ARN '" << subscriptionARN
<< "' was successful." << std::endl;
}
else {
std::cerr << "Error with TopicsAndQueues::Unsubscribe. "
<< outcome.GetError().GetMessage()
<< std::endl;
result = false;
}
}
}
printAsterisksLine();
if (!topicARN.empty() && askUser &&
askYesNoQuestion("Delete the SNS topic? (y/n) ")) {
// 11. Delete an SNS topic.
Aws::SNS::Model::DeleteTopicRequest request;
request.SetTopicArn(topicARN);
Aws::SNS::Model::DeleteTopicOutcome outcome = snsClient.DeleteTopic(request);
if (outcome.IsSuccess()) {
std::cout << "The topic with ARN '" << topicARN
<< "' was successfully deleted." << std::endl;
}
else {
std::cerr << "Error with TopicsAndQueues::DeleteTopicRequest. "
<< outcome.GetError().GetMessage()
<< std::endl;
result = false;
}
}
return result;
}
//! Create an IAM policy that gives an SQS queue permission to receive messages from an SNS topic.
/*!
\sa createPolicyForQueue()
\param queueARN: The SQS queue Amazon Resource Name (ARN).
\param topicARN: The SNS topic ARN.
\return Aws::String: The policy as JSON.
*/
Aws::String AwsDoc::TopicsAndQueues::createPolicyForQueue(const Aws::String &queueARN,
const Aws::String &topicARN) {
std::ostringstream policyStream;
policyStream << R"({
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "sns.amazonaws.com"
},
"Action": "sqs:SendMessage",
"Resource": ")" << queueARN << R"(",
"Condition": {
"ArnEquals": {
"aws:SourceArn": ")" << topicARN << R"("
}
}
}
]
})";
return policyStream.str();
}
```
+ For API details, see the following topics in *AWS SDK for C\$1\$1 API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/SdkForCpp/sqs-2012-11-05/CreateQueue)
+ [CreateTopic](https://docs.aws.amazon.com/goto/SdkForCpp/sns-2010-03-31/CreateTopic)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/SdkForCpp/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/SdkForCpp/sqs-2012-11-05/DeleteQueue)
+ [DeleteTopic](https://docs.aws.amazon.com/goto/SdkForCpp/sns-2010-03-31/DeleteTopic)
+ [GetQueueAttributes](https://docs.aws.amazon.com/goto/SdkForCpp/sqs-2012-11-05/GetQueueAttributes)
+ [Publish](https://docs.aws.amazon.com/goto/SdkForCpp/sns-2010-03-31/Publish)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/SdkForCpp/sqs-2012-11-05/ReceiveMessage)
+ [SetQueueAttributes](https://docs.aws.amazon.com/goto/SdkForCpp/sqs-2012-11-05/SetQueueAttributes)
+ [Subscribe](https://docs.aws.amazon.com/goto/SdkForCpp/sns-2010-03-31/Subscribe)
+ [Unsubscribe](https://docs.aws.amazon.com/goto/SdkForCpp/sns-2010-03-31/Unsubscribe)
------
#### [ Go ]
**SDK for Go V2**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/gov2/workflows/topics_and_queues#code-examples).
Run an interactive scenario at a command prompt.
```
import (
"context"
"encoding/json"
"fmt"
"log"
"strings"
"topics_and_queues/actions"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/sns"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
"github.com/awsdocs/aws-doc-sdk-examples/gov2/demotools"
)
const FIFO_SUFFIX = ".fifo"
const TONE_KEY = "tone"
var ToneChoices = []string{"cheerful", "funny", "serious", "sincere"}
// MessageBody is used to deserialize the body of a message from a JSON string.
type MessageBody struct {
Message string
}
// ScenarioRunner separates the steps of this scenario into individual functions so that
// they are simpler to read and understand.
type ScenarioRunner struct {
questioner demotools.IQuestioner
snsActor *actions.SnsActions
sqsActor *actions.SqsActions
}
func (runner ScenarioRunner) CreateTopic(ctx context.Context) (string, string, bool, bool) {
log.Println("SNS topics can be configured as FIFO (First-In-First-Out) or standard.\n" +
"FIFO topics deliver messages in order and support deduplication and message filtering.")
isFifoTopic := runner.questioner.AskBool("\nWould you like to work with FIFO topics? (y/n) ", "y")
contentBasedDeduplication := false
if isFifoTopic {
log.Println(strings.Repeat("-", 88))
log.Println("Because you have chosen a FIFO topic, deduplication is supported.\n" +
"Deduplication IDs are either set in the message or are automatically generated\n" +
"from content using a hash function. If a message is successfully published to\n" +
"an SNS FIFO topic, any message published and determined to have the same\n" +
"deduplication ID, within the five-minute deduplication interval, is accepted\n" +
"but not delivered. For more information about deduplication, see:\n" +
"\thttps://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html.")
contentBasedDeduplication = runner.questioner.AskBool(
"\nDo you want to use content-based deduplication instead of entering a deduplication ID? (y/n) ", "y")
}
log.Println(strings.Repeat("-", 88))
topicName := runner.questioner.Ask("Enter a name for your SNS topic. ")
if isFifoTopic {
topicName = fmt.Sprintf("%v%v", topicName, FIFO_SUFFIX)
log.Printf("Because you have selected a FIFO topic, '%v' must be appended to\n"+
"the topic name.", FIFO_SUFFIX)
}
topicArn, err := runner.snsActor.CreateTopic(ctx, topicName, isFifoTopic, contentBasedDeduplication)
if err != nil {
panic(err)
}
log.Printf("Your new topic with the name '%v' and Amazon Resource Name (ARN) \n"+
"'%v' has been created.", topicName, topicArn)
return topicName, topicArn, isFifoTopic, contentBasedDeduplication
}
func (runner ScenarioRunner) CreateQueue(ctx context.Context, ordinal string, isFifoTopic bool) (string, string) {
queueName := runner.questioner.Ask(fmt.Sprintf("Enter a name for the %v SQS queue. ", ordinal))
if isFifoTopic {
queueName = fmt.Sprintf("%v%v", queueName, FIFO_SUFFIX)
if ordinal == "first" {
log.Printf("Because you are creating a FIFO SQS queue, '%v' must "+
"be appended to the queue name.\n", FIFO_SUFFIX)
}
}
queueUrl, err := runner.sqsActor.CreateQueue(ctx, queueName, isFifoTopic)
if err != nil {
panic(err)
}
log.Printf("Your new SQS queue with the name '%v' and the queue URL "+
"'%v' has been created.", queueName, queueUrl)
return queueName, queueUrl
}
func (runner ScenarioRunner) SubscribeQueueToTopic(
ctx context.Context, queueName string, queueUrl string, topicName string, topicArn string, ordinal string,
isFifoTopic bool) (string, bool) {
queueArn, err := runner.sqsActor.GetQueueArn(ctx, queueUrl)
if err != nil {
panic(err)
}
log.Printf("The ARN of your queue is: %v.\n", queueArn)
err = runner.sqsActor.AttachSendMessagePolicy(ctx, queueUrl, queueArn, topicArn)
if err != nil {
panic(err)
}
log.Println("Attached an IAM policy to the queue so the SNS topic can send " +
"messages to it.")
log.Println(strings.Repeat("-", 88))
var filterPolicy map[string][]string
if isFifoTopic {
if ordinal == "first" {
log.Println("Subscriptions to a FIFO topic can have filters.\n" +
"If you add a filter to this subscription, then only the filtered messages\n" +
"will be received in the queue.\n" +
"For information about message filtering, see\n" +
"\thttps://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html\n" +
"For this example, you can filter messages by a \"tone\" attribute.")
}
wantFiltering := runner.questioner.AskBool(
fmt.Sprintf("Do you want to filter messages that are sent to \"%v\"\n"+
"from the %v topic? (y/n) ", queueName, topicName), "y")
if wantFiltering {
log.Println("You can filter messages by one or more of the following \"tone\" attributes.")
var toneSelections []string
askAboutTones := true
for askAboutTones {
toneIndex := runner.questioner.AskChoice(
"Enter the number of the tone you want to filter by:\n", ToneChoices)
toneSelections = append(toneSelections, ToneChoices[toneIndex])
askAboutTones = runner.questioner.AskBool("Do you want to add another tone to the filter? (y/n) ", "y")
}
log.Printf("Your subscription will be filtered to only pass the following tones: %v\n", toneSelections)
filterPolicy = map[string][]string{TONE_KEY: toneSelections}
}
}
subscriptionArn, err := runner.snsActor.SubscribeQueue(ctx, topicArn, queueArn, filterPolicy)
if err != nil {
panic(err)
}
log.Printf("The queue %v is now subscribed to the topic %v with the subscription ARN %v.\n",
queueName, topicName, subscriptionArn)
return subscriptionArn, filterPolicy != nil
}
func (runner ScenarioRunner) PublishMessages(ctx context.Context, topicArn string, isFifoTopic bool, contentBasedDeduplication bool, usingFilters bool) {
var message string
var groupId string
var dedupId string
var toneSelection string
publishMore := true
for publishMore {
groupId = ""
dedupId = ""
toneSelection = ""
message = runner.questioner.Ask("Enter a message to publish: ")
if isFifoTopic {
log.Println("Because you are using a FIFO topic, you must set a message group ID.\n" +
"All messages within the same group will be received in the order they were published.")
groupId = runner.questioner.Ask("Enter a message group ID: ")
if !contentBasedDeduplication {
log.Println("Because you are not using content-based deduplication,\n" +
"you must enter a deduplication ID.")
dedupId = runner.questioner.Ask("Enter a deduplication ID: ")
}
}
if usingFilters {
if runner.questioner.AskBool("Add a tone attribute so this message can be filtered? (y/n) ", "y") {
toneIndex := runner.questioner.AskChoice(
"Enter the number of the tone you want to filter by:\n", ToneChoices)
toneSelection = ToneChoices[toneIndex]
}
}
err := runner.snsActor.Publish(ctx, topicArn, message, groupId, dedupId, TONE_KEY, toneSelection)
if err != nil {
panic(err)
}
log.Println(("Your message was published."))
publishMore = runner.questioner.AskBool("Do you want to publish another messsage? (y/n) ", "y")
}
}
func (runner ScenarioRunner) PollForMessages(ctx context.Context, queueUrls []string) {
log.Println("Polling queues for messages...")
for _, queueUrl := range queueUrls {
var messages []types.Message
for {
currentMsgs, err := runner.sqsActor.GetMessages(ctx, queueUrl, 10, 1)
if err != nil {
panic(err)
}
if len(currentMsgs) == 0 {
break
}
messages = append(messages, currentMsgs...)
}
if len(messages) == 0 {
log.Printf("No messages were received by queue %v.\n", queueUrl)
} else if len(messages) == 1 {
log.Printf("One message was received by queue %v:\n", queueUrl)
} else {
log.Printf("%v messages were received by queue %v:\n", len(messages), queueUrl)
}
for msgIndex, message := range messages {
messageBody := MessageBody{}
err := json.Unmarshal([]byte(*message.Body), &messageBody)
if err != nil {
panic(err)
}
log.Printf("Message %v: %v\n", msgIndex+1, messageBody.Message)
}
if len(messages) > 0 {
log.Printf("Deleting %v messages from queue %v.\n", len(messages), queueUrl)
err := runner.sqsActor.DeleteMessages(ctx, queueUrl, messages)
if err != nil {
panic(err)
}
}
}
}
// RunTopicsAndQueuesScenario is an interactive example that shows you how to use the
// AWS SDK for Go to create and use Amazon SNS topics and Amazon SQS queues.
//
// 1. Create a topic (FIFO or non-FIFO).
// 2. Subscribe several queues to the topic with an option to apply a filter.
// 3. Publish messages to the topic.
// 4. Poll the queues for messages received.
// 5. Delete the topic and the queues.
//
// This example creates service clients from the specified sdkConfig so that
// you can replace it with a mocked or stubbed config for unit testing.
//
// It uses a questioner from the `demotools` package to get input during the example.
// This package can be found in the ..\..\demotools folder of this repo.
func RunTopicsAndQueuesScenario(
ctx context.Context, sdkConfig aws.Config, questioner demotools.IQuestioner) {
resources := Resources{}
defer func() {
if r := recover(); r != nil {
log.Println("Something went wrong with the demo.\n" +
"Cleaning up any resources that were created...")
resources.Cleanup(ctx)
}
}()
queueCount := 2
log.Println(strings.Repeat("-", 88))
log.Printf("Welcome to messaging with topics and queues.\n\n"+
"In this scenario, you will create an SNS topic and subscribe %v SQS queues to the\n"+
"topic. You can select from several options for configuring the topic and the\n"+
"subscriptions for the queues. You can then post to the topic and see the results\n"+
"in the queues.\n", queueCount)
log.Println(strings.Repeat("-", 88))
runner := ScenarioRunner{
questioner: questioner,
snsActor: &actions.SnsActions{SnsClient: sns.NewFromConfig(sdkConfig)},
sqsActor: &actions.SqsActions{SqsClient: sqs.NewFromConfig(sdkConfig)},
}
resources.snsActor = runner.snsActor
resources.sqsActor = runner.sqsActor
topicName, topicArn, isFifoTopic, contentBasedDeduplication := runner.CreateTopic(ctx)
resources.topicArn = topicArn
log.Println(strings.Repeat("-", 88))
log.Printf("Now you will create %v SQS queues and subscribe them to the topic.\n", queueCount)
ordinals := []string{"first", "next"}
usingFilters := false
for _, ordinal := range ordinals {
queueName, queueUrl := runner.CreateQueue(ctx, ordinal, isFifoTopic)
resources.queueUrls = append(resources.queueUrls, queueUrl)
_, filtering := runner.SubscribeQueueToTopic(ctx, queueName, queueUrl, topicName, topicArn, ordinal, isFifoTopic)
usingFilters = usingFilters || filtering
}
log.Println(strings.Repeat("-", 88))
runner.PublishMessages(ctx, topicArn, isFifoTopic, contentBasedDeduplication, usingFilters)
log.Println(strings.Repeat("-", 88))
runner.PollForMessages(ctx, resources.queueUrls)
log.Println(strings.Repeat("-", 88))
wantCleanup := questioner.AskBool("Do you want to remove all AWS resources created for this scenario? (y/n) ", "y")
if wantCleanup {
log.Println("Cleaning up resources...")
resources.Cleanup(ctx)
}
log.Println(strings.Repeat("-", 88))
log.Println("Thanks for watching!")
log.Println(strings.Repeat("-", 88))
}
```
Define a struct that wraps Amazon SNS actions used in this example.
```
import (
"context"
"encoding/json"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/sns"
"github.com/aws/aws-sdk-go-v2/service/sns/types"
)
// SnsActions encapsulates the Amazon Simple Notification Service (Amazon SNS) actions
// used in the examples.
type SnsActions struct {
SnsClient *sns.Client
}
// CreateTopic creates an Amazon SNS topic with the specified name. You can optionally
// specify that the topic is created as a FIFO topic and whether it uses content-based
// deduplication instead of ID-based deduplication.
func (actor SnsActions) CreateTopic(ctx context.Context, topicName string, isFifoTopic bool, contentBasedDeduplication bool) (string, error) {
var topicArn string
topicAttributes := map[string]string{}
if isFifoTopic {
topicAttributes["FifoTopic"] = "true"
}
if contentBasedDeduplication {
topicAttributes["ContentBasedDeduplication"] = "true"
}
topic, err := actor.SnsClient.CreateTopic(ctx, &sns.CreateTopicInput{
Name: aws.String(topicName),
Attributes: topicAttributes,
})
if err != nil {
log.Printf("Couldn't create topic %v. Here's why: %v\n", topicName, err)
} else {
topicArn = *topic.TopicArn
}
return topicArn, err
}
// DeleteTopic delete an Amazon SNS topic.
func (actor SnsActions) DeleteTopic(ctx context.Context, topicArn string) error {
_, err := actor.SnsClient.DeleteTopic(ctx, &sns.DeleteTopicInput{
TopicArn: aws.String(topicArn)})
if err != nil {
log.Printf("Couldn't delete topic %v. Here's why: %v\n", topicArn, err)
}
return err
}
// SubscribeQueue subscribes an Amazon Simple Queue Service (Amazon SQS) queue to an
// Amazon SNS topic. When filterMap is not nil, it is used to specify a filter policy
// so that messages are only sent to the queue when the message has the specified attributes.
func (actor SnsActions) SubscribeQueue(ctx context.Context, topicArn string, queueArn string, filterMap map[string][]string) (string, error) {
var subscriptionArn string
var attributes map[string]string
if filterMap != nil {
filterBytes, err := json.Marshal(filterMap)
if err != nil {
log.Printf("Couldn't create filter policy, here's why: %v\n", err)
return "", err
}
attributes = map[string]string{"FilterPolicy": string(filterBytes)}
}
output, err := actor.SnsClient.Subscribe(ctx, &sns.SubscribeInput{
Protocol: aws.String("sqs"),
TopicArn: aws.String(topicArn),
Attributes: attributes,
Endpoint: aws.String(queueArn),
ReturnSubscriptionArn: true,
})
if err != nil {
log.Printf("Couldn't susbscribe queue %v to topic %v. Here's why: %v\n",
queueArn, topicArn, err)
} else {
subscriptionArn = *output.SubscriptionArn
}
return subscriptionArn, err
}
// Publish publishes a message to an Amazon SNS topic. The message is then sent to all
// subscribers. When the topic is a FIFO topic, the message must also contain a group ID
// and, when ID-based deduplication is used, a deduplication ID. An optional key-value
// filter attribute can be specified so that the message can be filtered according to
// a filter policy.
func (actor SnsActions) Publish(ctx context.Context, topicArn string, message string, groupId string, dedupId string, filterKey string, filterValue string) error {
publishInput := sns.PublishInput{TopicArn: aws.String(topicArn), Message: aws.String(message)}
if groupId != "" {
publishInput.MessageGroupId = aws.String(groupId)
}
if dedupId != "" {
publishInput.MessageDeduplicationId = aws.String(dedupId)
}
if filterKey != "" && filterValue != "" {
publishInput.MessageAttributes = map[string]types.MessageAttributeValue{
filterKey: {DataType: aws.String("String"), StringValue: aws.String(filterValue)},
}
}
_, err := actor.SnsClient.Publish(ctx, &publishInput)
if err != nil {
log.Printf("Couldn't publish message to topic %v. Here's why: %v", topicArn, err)
}
return err
}
```
Define a struct that wraps Amazon SQS actions used in this example.
```
import (
"context"
"encoding/json"
"fmt"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
)
// SqsActions encapsulates the Amazon Simple Queue Service (Amazon SQS) actions
// used in the examples.
type SqsActions struct {
SqsClient *sqs.Client
}
// CreateQueue creates an Amazon SQS queue with the specified name. You can specify
// whether the queue is created as a FIFO queue.
func (actor SqsActions) CreateQueue(ctx context.Context, queueName string, isFifoQueue bool) (string, error) {
var queueUrl string
queueAttributes := map[string]string{}
if isFifoQueue {
queueAttributes["FifoQueue"] = "true"
}
queue, err := actor.SqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{
QueueName: aws.String(queueName),
Attributes: queueAttributes,
})
if err != nil {
log.Printf("Couldn't create queue %v. Here's why: %v\n", queueName, err)
} else {
queueUrl = *queue.QueueUrl
}
return queueUrl, err
}
// GetQueueArn uses the GetQueueAttributes action to get the Amazon Resource Name (ARN)
// of an Amazon SQS queue.
func (actor SqsActions) GetQueueArn(ctx context.Context, queueUrl string) (string, error) {
var queueArn string
arnAttributeName := types.QueueAttributeNameQueueArn
attribute, err := actor.SqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
QueueUrl: aws.String(queueUrl),
AttributeNames: []types.QueueAttributeName{arnAttributeName},
})
if err != nil {
log.Printf("Couldn't get ARN for queue %v. Here's why: %v\n", queueUrl, err)
} else {
queueArn = attribute.Attributes[string(arnAttributeName)]
}
return queueArn, err
}
// AttachSendMessagePolicy uses the SetQueueAttributes action to attach a policy to an
// Amazon SQS queue that allows the specified Amazon SNS topic to send messages to the
// queue.
func (actor SqsActions) AttachSendMessagePolicy(ctx context.Context, queueUrl string, queueArn string, topicArn string) error {
policyDoc := PolicyDocument{
Version: "2012-10-17",
Statement: []PolicyStatement{{
Effect: "Allow",
Action: "sqs:SendMessage",
Principal: map[string]string{"Service": "sns.amazonaws.com"},
Resource: aws.String(queueArn),
Condition: PolicyCondition{"ArnEquals": map[string]string{"aws:SourceArn": topicArn}},
}},
}
policyBytes, err := json.Marshal(policyDoc)
if err != nil {
log.Printf("Couldn't create policy document. Here's why: %v\n", err)
return err
}
_, err = actor.SqsClient.SetQueueAttributes(ctx, &sqs.SetQueueAttributesInput{
Attributes: map[string]string{
string(types.QueueAttributeNamePolicy): string(policyBytes),
},
QueueUrl: aws.String(queueUrl),
})
if err != nil {
log.Printf("Couldn't set send message policy on queue %v. Here's why: %v\n", queueUrl, err)
}
return err
}
// PolicyDocument defines a policy document as a Go struct that can be serialized
// to JSON.
type PolicyDocument struct {
Version string
Statement []PolicyStatement
}
// PolicyStatement defines a statement in a policy document.
type PolicyStatement struct {
Effect string
Action string
Principal map[string]string `json:",omitempty"`
Resource *string `json:",omitempty"`
Condition PolicyCondition `json:",omitempty"`
}
// PolicyCondition defines a condition in a policy.
type PolicyCondition map[string]map[string]string
// GetMessages uses the ReceiveMessage action to get messages from an Amazon SQS queue.
func (actor SqsActions) GetMessages(ctx context.Context, queueUrl string, maxMessages int32, waitTime int32) ([]types.Message, error) {
var messages []types.Message
result, err := actor.SqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
QueueUrl: aws.String(queueUrl),
MaxNumberOfMessages: maxMessages,
WaitTimeSeconds: waitTime,
})
if err != nil {
log.Printf("Couldn't get messages from queue %v. Here's why: %v\n", queueUrl, err)
} else {
messages = result.Messages
}
return messages, err
}
// DeleteMessages uses the DeleteMessageBatch action to delete a batch of messages from
// an Amazon SQS queue.
func (actor SqsActions) DeleteMessages(ctx context.Context, queueUrl string, messages []types.Message) error {
entries := make([]types.DeleteMessageBatchRequestEntry, len(messages))
for msgIndex := range messages {
entries[msgIndex].Id = aws.String(fmt.Sprintf("%v", msgIndex))
entries[msgIndex].ReceiptHandle = messages[msgIndex].ReceiptHandle
}
_, err := actor.SqsClient.DeleteMessageBatch(ctx, &sqs.DeleteMessageBatchInput{
Entries: entries,
QueueUrl: aws.String(queueUrl),
})
if err != nil {
log.Printf("Couldn't delete messages from queue %v. Here's why: %v\n", queueUrl, err)
}
return err
}
// DeleteQueue deletes an Amazon SQS queue.
func (actor SqsActions) DeleteQueue(ctx context.Context, queueUrl string) error {
_, err := actor.SqsClient.DeleteQueue(ctx, &sqs.DeleteQueueInput{
QueueUrl: aws.String(queueUrl)})
if err != nil {
log.Printf("Couldn't delete queue %v. Here's why: %v\n", queueUrl, err)
}
return err
}
```
Clean up resources.
```
import (
"context"
"fmt"
"log"
"topics_and_queues/actions"
)
// Resources keeps track of AWS resources created during an example and handles
// cleanup when the example finishes.
type Resources struct {
topicArn string
queueUrls []string
snsActor *actions.SnsActions
sqsActor *actions.SqsActions
}
// Cleanup deletes all AWS resources created during an example.
func (resources Resources) Cleanup(ctx context.Context) {
defer func() {
if r := recover(); r != nil {
fmt.Println("Something went wrong during cleanup. Use the AWS Management Console\n" +
"to remove any remaining resources that were created for this scenario.")
}
}()
var err error
if resources.topicArn != "" {
log.Printf("Deleting topic %v.\n", resources.topicArn)
err = resources.snsActor.DeleteTopic(ctx, resources.topicArn)
if err != nil {
panic(err)
}
}
for _, queueUrl := range resources.queueUrls {
log.Printf("Deleting queue %v.\n", queueUrl)
err = resources.sqsActor.DeleteQueue(ctx, queueUrl)
if err != nil {
panic(err)
}
}
}
```
+ For API details, see the following topics in *AWS SDK for Go API Reference*.
+ [CreateQueue](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sqs#Client.CreateQueue)
+ [CreateTopic](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sns#Client.CreateTopic)
+ [DeleteMessageBatch](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sqs#Client.DeleteMessageBatch)
+ [DeleteQueue](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sqs#Client.DeleteQueue)
+ [DeleteTopic](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sns#Client.DeleteTopic)
+ [GetQueueAttributes](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sqs#Client.GetQueueAttributes)
+ [Publish](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sns#Client.Publish)
+ [ReceiveMessage](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sqs#Client.ReceiveMessage)
+ [SetQueueAttributes](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sqs#Client.SetQueueAttributes)
+ [Subscribe](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sns#Client.Subscribe)
+ [Unsubscribe](https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/sns#Client.Unsubscribe)
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/usecases/topics_and_queues#code-examples).
```
package com.example.sns;
import software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sns.SnsClient;
import software.amazon.awssdk.services.sns.model.CreateTopicRequest;
import software.amazon.awssdk.services.sns.model.CreateTopicResponse;
import software.amazon.awssdk.services.sns.model.DeleteTopicRequest;
import software.amazon.awssdk.services.sns.model.DeleteTopicResponse;
import software.amazon.awssdk.services.sns.model.MessageAttributeValue;
import software.amazon.awssdk.services.sns.model.PublishRequest;
import software.amazon.awssdk.services.sns.model.PublishResponse;
import software.amazon.awssdk.services.sns.model.SetSubscriptionAttributesRequest;
import software.amazon.awssdk.services.sns.model.SnsException;
import software.amazon.awssdk.services.sns.model.SubscribeRequest;
import software.amazon.awssdk.services.sns.model.SubscribeResponse;
import software.amazon.awssdk.services.sns.model.UnsubscribeRequest;
import software.amazon.awssdk.services.sns.model.UnsubscribeResponse;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.CreateQueueRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
import software.amazon.awssdk.services.sqs.model.DeleteQueueRequest;
import software.amazon.awssdk.services.sqs.model.GetQueueAttributesRequest;
import software.amazon.awssdk.services.sqs.model.GetQueueAttributesResponse;
import software.amazon.awssdk.services.sqs.model.GetQueueUrlRequest;
import software.amazon.awssdk.services.sqs.model.GetQueueUrlResponse;
import software.amazon.awssdk.services.sqs.model.Message;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.SetQueueAttributesRequest;
import software.amazon.awssdk.services.sqs.model.SqsException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.gson.JsonPrimitive;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*
* This Java example performs these tasks:
*
* 1. Gives the user three options to choose from.
* 2. Creates an Amazon Simple Notification Service (Amazon SNS) topic.
* 3. Creates an Amazon Simple Queue Service (Amazon SQS) queue.
* 4. Gets the SQS queue Amazon Resource Name (ARN) attribute.
* 5. Attaches an AWS Identity and Access Management (IAM) policy to the queue.
* 6. Subscribes to the SQS queue.
* 7. Publishes a message to the topic.
* 8. Displays the messages.
* 9. Deletes the received message.
* 10. Unsubscribes from the topic.
* 11. Deletes the SNS topic.
*/
public class SNSWorkflow {
public static final String DASHES = new String(new char[80]).replace("\0", "-");
public static void main(String[] args) {
final String usage = "\n" +
"Usage:\n" +
" \n\n" +
"Where:\n" +
" accountId - Your AWS account Id value.";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
SnsClient snsClient = SnsClient.builder()
.region(Region.US_EAST_1)
.credentialsProvider(EnvironmentVariableCredentialsProvider.create())
.build();
SqsClient sqsClient = SqsClient.builder()
.region(Region.US_EAST_1)
.credentialsProvider(EnvironmentVariableCredentialsProvider.create())
.build();
Scanner in = new Scanner(System.in);
String accountId = args[0];
String useFIFO;
String duplication = "n";
String topicName;
String deduplicationID = null;
String groupId = null;
String topicArn;
String sqsQueueName;
String sqsQueueUrl;
String sqsQueueArn;
String subscriptionArn;
boolean selectFIFO = false;
String message;
List messageList;
List filterList = new ArrayList<>();
String msgAttValue = "";
System.out.println(DASHES);
System.out.println("Welcome to messaging with topics and queues.");
System.out.println("In this scenario, you will create an SNS topic and subscribe an SQS queue to the topic.\n" +
"You can select from several options for configuring the topic and the subscriptions for the queue.\n" +
"You can then post to the topic and see the results in the queue.");
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("SNS topics can be configured as FIFO (First-In-First-Out).\n" +
"FIFO topics deliver messages in order and support deduplication and message filtering.\n" +
"Would you like to work with FIFO topics? (y/n)");
useFIFO = in.nextLine();
if (useFIFO.compareTo("y") == 0) {
selectFIFO = true;
System.out.println("You have selected FIFO");
System.out.println(" Because you have chosen a FIFO topic, deduplication is supported.\n" +
" Deduplication IDs are either set in the message or automatically generated from content using a hash function.\n"
+
" If a message is successfully published to an SNS FIFO topic, any message published and determined to have the same deduplication ID,\n"
+
" within the five-minute deduplication interval, is accepted but not delivered.\n" +
" For more information about deduplication, see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html.");
System.out.println(
"Would you like to use content-based deduplication instead of entering a deduplication ID? (y/n)");
duplication = in.nextLine();
if (duplication.compareTo("y") == 0) {
System.out.println("Please enter a group id value");
groupId = in.nextLine();
} else {
System.out.println("Please enter deduplication Id value");
deduplicationID = in.nextLine();
System.out.println("Please enter a group id value");
groupId = in.nextLine();
}
}
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("2. Create a topic.");
System.out.println("Enter a name for your SNS topic.");
topicName = in.nextLine();
if (selectFIFO) {
System.out.println("Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.");
topicName = topicName + ".fifo";
System.out.println("The name of the topic is " + topicName);
topicArn = createFIFO(snsClient, topicName, duplication);
System.out.println("The ARN of the FIFO topic is " + topicArn);
} else {
System.out.println("The name of the topic is " + topicName);
topicArn = createSNSTopic(snsClient, topicName);
System.out.println("The ARN of the non-FIFO topic is " + topicArn);
}
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("3. Create an SQS queue.");
System.out.println("Enter a name for your SQS queue.");
sqsQueueName = in.nextLine();
if (selectFIFO) {
sqsQueueName = sqsQueueName + ".fifo";
}
sqsQueueUrl = createQueue(sqsClient, sqsQueueName, selectFIFO);
System.out.println("The queue URL is " + sqsQueueUrl);
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("4. Get the SQS queue ARN attribute.");
sqsQueueArn = getSQSQueueAttrs(sqsClient, sqsQueueUrl);
System.out.println("The ARN of the new queue is " + sqsQueueArn);
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("5. Attach an IAM policy to the queue.");
// Define the policy to use. Make sure that you change the REGION if you are
// running this code
// in a different region.
String policy = """
{
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "sns.amazonaws.com"
},
"Action": "sqs:SendMessage",
"Resource": "arn:aws:sqs:us-east-1:%s:%s",
"Condition": {
"ArnEquals": {
"aws:SourceArn": "arn:aws:sns:us-east-1:%s:%s"
}
}
}
]
}
""".formatted(accountId, sqsQueueName, accountId, topicName);
setQueueAttr(sqsClient, sqsQueueUrl, policy);
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("6. Subscribe to the SQS queue.");
if (selectFIFO) {
System.out.println(
"If you add a filter to this subscription, then only the filtered messages will be received in the queue.\n"
+
"For information about message filtering, see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html\n"
+
"For this example, you can filter messages by a \"tone\" attribute.");
System.out.println("Would you like to filter messages for " + sqsQueueName + "'s subscription to the topic "
+ topicName + "? (y/n)");
String filterAns = in.nextLine();
if (filterAns.compareTo("y") == 0) {
boolean moreAns = false;
System.out.println("You can filter messages by one or more of the following \"tone\" attributes.");
System.out.println("1. cheerful");
System.out.println("2. funny");
System.out.println("3. serious");
System.out.println("4. sincere");
while (!moreAns) {
System.out.println("Select a number or choose 0 to end.");
String ans = in.nextLine();
switch (ans) {
case "1":
filterList.add("cheerful");
break;
case "2":
filterList.add("funny");
break;
case "3":
filterList.add("serious");
break;
case "4":
filterList.add("sincere");
break;
default:
moreAns = true;
break;
}
}
}
}
subscriptionArn = subQueue(snsClient, topicArn, sqsQueueArn, filterList);
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("7. Publish a message to the topic.");
if (selectFIFO) {
System.out.println("Would you like to add an attribute to this message? (y/n)");
String msgAns = in.nextLine();
if (msgAns.compareTo("y") == 0) {
System.out.println("You can filter messages by one or more of the following \"tone\" attributes.");
System.out.println("1. cheerful");
System.out.println("2. funny");
System.out.println("3. serious");
System.out.println("4. sincere");
System.out.println("Select a number or choose 0 to end.");
String ans = in.nextLine();
switch (ans) {
case "1":
msgAttValue = "cheerful";
break;
case "2":
msgAttValue = "funny";
break;
case "3":
msgAttValue = "serious";
break;
default:
msgAttValue = "sincere";
break;
}
System.out.println("Selected value is " + msgAttValue);
}
System.out.println("Enter a message.");
message = in.nextLine();
pubMessageFIFO(snsClient, message, topicArn, msgAttValue, duplication, groupId, deduplicationID);
} else {
System.out.println("Enter a message.");
message = in.nextLine();
pubMessage(snsClient, message, topicArn);
}
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("8. Display the message. Press any key to continue.");
in.nextLine();
messageList = receiveMessages(sqsClient, sqsQueueUrl, msgAttValue);
for (Message mes : messageList) {
System.out.println("Message Id: " + mes.messageId());
System.out.println("Full Message: " + mes.body());
}
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("9. Delete the received message. Press any key to continue.");
in.nextLine();
deleteMessages(sqsClient, sqsQueueUrl, messageList);
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("10. Unsubscribe from the topic and delete the queue. Press any key to continue.");
in.nextLine();
unSub(snsClient, subscriptionArn);
deleteSQSQueue(sqsClient, sqsQueueName);
System.out.println(DASHES);
System.out.println(DASHES);
System.out.println("11. Delete the topic. Press any key to continue.");
in.nextLine();
deleteSNSTopic(snsClient, topicArn);
System.out.println(DASHES);
System.out.println("The SNS/SQS workflow has completed successfully.");
System.out.println(DASHES);
}
public static void deleteSNSTopic(SnsClient snsClient, String topicArn) {
try {
DeleteTopicRequest request = DeleteTopicRequest.builder()
.topicArn(topicArn)
.build();
DeleteTopicResponse result = snsClient.deleteTopic(request);
System.out.println("Status was " + result.sdkHttpResponse().statusCode());
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
public static void deleteSQSQueue(SqsClient sqsClient, String queueName) {
try {
GetQueueUrlRequest getQueueRequest = GetQueueUrlRequest.builder()
.queueName(queueName)
.build();
String queueUrl = sqsClient.getQueueUrl(getQueueRequest).queueUrl();
DeleteQueueRequest deleteQueueRequest = DeleteQueueRequest.builder()
.queueUrl(queueUrl)
.build();
sqsClient.deleteQueue(deleteQueueRequest);
System.out.println(queueName + " was successfully deleted.");
} catch (SqsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
public static void unSub(SnsClient snsClient, String subscriptionArn) {
try {
UnsubscribeRequest request = UnsubscribeRequest.builder()
.subscriptionArn(subscriptionArn)
.build();
UnsubscribeResponse result = snsClient.unsubscribe(request);
System.out.println("Status was " + result.sdkHttpResponse().statusCode()
+ "\nSubscription was removed for " + request.subscriptionArn());
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
public static void deleteMessages(SqsClient sqsClient, String queueUrl, List messages) {
try {
List entries = new ArrayList<>();
for (Message msg : messages) {
DeleteMessageBatchRequestEntry entry = DeleteMessageBatchRequestEntry.builder()
.id(msg.messageId())
.build();
entries.add(entry);
}
DeleteMessageBatchRequest deleteMessageBatchRequest = DeleteMessageBatchRequest.builder()
.queueUrl(queueUrl)
.entries(entries)
.build();
sqsClient.deleteMessageBatch(deleteMessageBatchRequest);
System.out.println("The batch delete of messages was successful");
} catch (SqsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
public static List receiveMessages(SqsClient sqsClient, String queueUrl, String msgAttValue) {
try {
if (msgAttValue.isEmpty()) {
ReceiveMessageRequest receiveMessageRequest = ReceiveMessageRequest.builder()
.queueUrl(queueUrl)
.maxNumberOfMessages(5)
.build();
return sqsClient.receiveMessage(receiveMessageRequest).messages();
} else {
// We know there are filters on the message.
ReceiveMessageRequest receiveRequest = ReceiveMessageRequest.builder()
.queueUrl(queueUrl)
.messageAttributeNames(msgAttValue) // Include other message attributes if needed.
.maxNumberOfMessages(5)
.build();
return sqsClient.receiveMessage(receiveRequest).messages();
}
} catch (SqsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
return null;
}
public static void pubMessage(SnsClient snsClient, String message, String topicArn) {
try {
PublishRequest request = PublishRequest.builder()
.message(message)
.topicArn(topicArn)
.build();
PublishResponse result = snsClient.publish(request);
System.out
.println(result.messageId() + " Message sent. Status is " + result.sdkHttpResponse().statusCode());
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
public static void pubMessageFIFO(SnsClient snsClient,
String message,
String topicArn,
String msgAttValue,
String duplication,
String groupId,
String deduplicationID) {
try {
PublishRequest request;
// Means the user did not choose to use a message attribute.
if (msgAttValue.isEmpty()) {
if (duplication.compareTo("y") == 0) {
request = PublishRequest.builder()
.message(message)
.messageGroupId(groupId)
.topicArn(topicArn)
.build();
} else {
request = PublishRequest.builder()
.message(message)
.messageDeduplicationId(deduplicationID)
.messageGroupId(groupId)
.topicArn(topicArn)
.build();
}
} else {
Map messageAttributes = new HashMap<>();
messageAttributes.put(msgAttValue, MessageAttributeValue.builder()
.dataType("String")
.stringValue("true")
.build());
if (duplication.compareTo("y") == 0) {
request = PublishRequest.builder()
.message(message)
.messageGroupId(groupId)
.topicArn(topicArn)
.build();
} else {
// Create a publish request with the message and attributes.
request = PublishRequest.builder()
.topicArn(topicArn)
.message(message)
.messageDeduplicationId(deduplicationID)
.messageGroupId(groupId)
.messageAttributes(messageAttributes)
.build();
}
}
// Publish the message to the topic.
PublishResponse result = snsClient.publish(request);
System.out
.println(result.messageId() + " Message sent. Status was " + result.sdkHttpResponse().statusCode());
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
// Subscribe to the SQS queue.
public static String subQueue(SnsClient snsClient, String topicArn, String queueArn, List filterList) {
try {
SubscribeRequest request;
if (filterList.isEmpty()) {
// No filter subscription is added.
request = SubscribeRequest.builder()
.protocol("sqs")
.endpoint(queueArn)
.returnSubscriptionArn(true)
.topicArn(topicArn)
.build();
SubscribeResponse result = snsClient.subscribe(request);
System.out.println("The queue " + queueArn + " has been subscribed to the topic " + topicArn + "\n" +
"with the subscription ARN " + result.subscriptionArn());
return result.subscriptionArn();
} else {
request = SubscribeRequest.builder()
.protocol("sqs")
.endpoint(queueArn)
.returnSubscriptionArn(true)
.topicArn(topicArn)
.build();
SubscribeResponse result = snsClient.subscribe(request);
System.out.println("The queue " + queueArn + " has been subscribed to the topic " + topicArn + "\n" +
"with the subscription ARN " + result.subscriptionArn());
String attributeName = "FilterPolicy";
Gson gson = new Gson();
String jsonString = "{\"tone\": []}";
JsonObject jsonObject = gson.fromJson(jsonString, JsonObject.class);
JsonArray toneArray = jsonObject.getAsJsonArray("tone");
for (String value : filterList) {
toneArray.add(new JsonPrimitive(value));
}
String updatedJsonString = gson.toJson(jsonObject);
System.out.println(updatedJsonString);
SetSubscriptionAttributesRequest attRequest = SetSubscriptionAttributesRequest.builder()
.subscriptionArn(result.subscriptionArn())
.attributeName(attributeName)
.attributeValue(updatedJsonString)
.build();
snsClient.setSubscriptionAttributes(attRequest);
return result.subscriptionArn();
}
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
return "";
}
// Attach a policy to the queue.
public static void setQueueAttr(SqsClient sqsClient, String queueUrl, String policy) {
try {
Map attrMap = new HashMap<>();
attrMap.put(QueueAttributeName.POLICY, policy);
SetQueueAttributesRequest attributesRequest = SetQueueAttributesRequest.builder()
.queueUrl(queueUrl)
.attributes(attrMap)
.build();
sqsClient.setQueueAttributes(attributesRequest);
System.out.println("The policy has been successfully attached.");
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
}
public static String getSQSQueueAttrs(SqsClient sqsClient, String queueUrl) {
// Specify the attributes to retrieve.
List atts = new ArrayList<>();
atts.add(QueueAttributeName.QUEUE_ARN);
GetQueueAttributesRequest attributesRequest = GetQueueAttributesRequest.builder()
.queueUrl(queueUrl)
.attributeNames(atts)
.build();
GetQueueAttributesResponse response = sqsClient.getQueueAttributes(attributesRequest);
Map queueAtts = response.attributesAsStrings();
for (Map.Entry queueAtt : queueAtts.entrySet())
return queueAtt.getValue();
return "";
}
public static String createQueue(SqsClient sqsClient, String queueName, Boolean selectFIFO) {
try {
System.out.println("\nCreate Queue");
if (selectFIFO) {
Map attrs = new HashMap<>();
attrs.put(QueueAttributeName.FIFO_QUEUE, "true");
CreateQueueRequest createQueueRequest = CreateQueueRequest.builder()
.queueName(queueName)
.attributes(attrs)
.build();
sqsClient.createQueue(createQueueRequest);
System.out.println("\nGet queue url");
GetQueueUrlResponse getQueueUrlResponse = sqsClient
.getQueueUrl(GetQueueUrlRequest.builder().queueName(queueName).build());
return getQueueUrlResponse.queueUrl();
} else {
CreateQueueRequest createQueueRequest = CreateQueueRequest.builder()
.queueName(queueName)
.build();
sqsClient.createQueue(createQueueRequest);
System.out.println("\nGet queue url");
GetQueueUrlResponse getQueueUrlResponse = sqsClient
.getQueueUrl(GetQueueUrlRequest.builder().queueName(queueName).build());
return getQueueUrlResponse.queueUrl();
}
} catch (SqsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
return "";
}
public static String createSNSTopic(SnsClient snsClient, String topicName) {
CreateTopicResponse result;
try {
CreateTopicRequest request = CreateTopicRequest.builder()
.name(topicName)
.build();
result = snsClient.createTopic(request);
return result.topicArn();
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
return "";
}
public static String createFIFO(SnsClient snsClient, String topicName, String duplication) {
try {
// Create a FIFO topic by using the SNS service client.
Map topicAttributes = new HashMap<>();
if (duplication.compareTo("n") == 0) {
topicAttributes.put("FifoTopic", "true");
topicAttributes.put("ContentBasedDeduplication", "false");
} else {
topicAttributes.put("FifoTopic", "true");
topicAttributes.put("ContentBasedDeduplication", "true");
}
CreateTopicRequest topicRequest = CreateTopicRequest.builder()
.name(topicName)
.attributes(topicAttributes)
.build();
CreateTopicResponse response = snsClient.createTopic(topicRequest);
return response.topicArn();
} catch (SnsException e) {
System.err.println(e.awsErrorDetails().errorMessage());
System.exit(1);
}
return "";
}
}
```
+ For API details, see the following topics in *AWS SDK for Java 2.x API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/CreateQueue)
+ [CreateTopic](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/CreateTopic)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteQueue)
+ [DeleteTopic](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/DeleteTopic)
+ [GetQueueAttributes](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/GetQueueAttributes)
+ [Publish](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/Publish)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/ReceiveMessage)
+ [SetQueueAttributes](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/SetQueueAttributes)
+ [Subscribe](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/Subscribe)
+ [Unsubscribe](https://docs.aws.amazon.com/goto/SdkForJavaV2/sns-2010-03-31/Unsubscribe)
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/cross-services/wkflw-topics-queues#code-examples).
This is the entry point for this scenario.
```
import { SNSClient } from "@aws-sdk/client-sns";
import { SQSClient } from "@aws-sdk/client-sqs";
import { TopicsQueuesWkflw } from "./TopicsQueuesWkflw.js";
import { Prompter } from "@aws-doc-sdk-examples/lib/prompter.js";
export const startSnsWorkflow = () => {
const snsClient = new SNSClient({});
const sqsClient = new SQSClient({});
const prompter = new Prompter();
const logger = console;
const wkflw = new TopicsQueuesWkflw(snsClient, sqsClient, prompter, logger);
wkflw.start();
};
```
The preceding code provides the necessary dependencies and starts the scenario. The next section contains the bulk of the example.
```
const toneChoices = [
{ name: "cheerful", value: "cheerful" },
{ name: "funny", value: "funny" },
{ name: "serious", value: "serious" },
{ name: "sincere", value: "sincere" },
];
export class TopicsQueuesWkflw {
// SNS topic is configured as First-In-First-Out
isFifo = true;
// Automatic content-based deduplication is enabled.
autoDedup = false;
snsClient;
sqsClient;
topicName;
topicArn;
subscriptionArns = [];
/**
* @type {{ queueName: string, queueArn: string, queueUrl: string, policy?: string }[]}
*/
queues = [];
prompter;
/**
* @param {import('@aws-sdk/client-sns').SNSClient} snsClient
* @param {import('@aws-sdk/client-sqs').SQSClient} sqsClient
* @param {import('../../libs/prompter.js').Prompter} prompter
* @param {import('../../libs/logger.js').Logger} logger
*/
constructor(snsClient, sqsClient, prompter, logger) {
this.snsClient = snsClient;
this.sqsClient = sqsClient;
this.prompter = prompter;
this.logger = logger;
}
async welcome() {
await this.logger.log(MESSAGES.description);
}
async confirmFifo() {
await this.logger.log(MESSAGES.snsFifoDescription);
this.isFifo = await this.prompter.confirm({
message: MESSAGES.snsFifoPrompt,
});
if (this.isFifo) {
this.logger.logSeparator(MESSAGES.headerDedup);
await this.logger.log(MESSAGES.deduplicationNotice);
await this.logger.log(MESSAGES.deduplicationDescription);
this.autoDedup = await this.prompter.confirm({
message: MESSAGES.deduplicationPrompt,
});
}
}
async createTopic() {
await this.logger.log(MESSAGES.creatingTopics);
this.topicName = await this.prompter.input({
message: MESSAGES.topicNamePrompt,
});
if (this.isFifo) {
this.topicName += ".fifo";
this.logger.logSeparator(MESSAGES.headerFifoNaming);
await this.logger.log(MESSAGES.appendFifoNotice);
}
const response = await this.snsClient.send(
new CreateTopicCommand({
Name: this.topicName,
Attributes: {
FifoTopic: this.isFifo ? "true" : "false",
...(this.autoDedup ? { ContentBasedDeduplication: "true" } : {}),
},
}),
);
this.topicArn = response.TopicArn;
await this.logger.log(
MESSAGES.topicCreatedNotice
.replace("${TOPIC_NAME}", this.topicName)
.replace("${TOPIC_ARN}", this.topicArn),
);
}
async createQueues() {
await this.logger.log(MESSAGES.createQueuesNotice);
// Increase this number to add more queues.
const maxQueues = 2;
for (let i = 0; i < maxQueues; i++) {
await this.logger.log(MESSAGES.queueCount.replace("${COUNT}", i + 1));
let queueName = await this.prompter.input({
message: MESSAGES.queueNamePrompt.replace(
"${EXAMPLE_NAME}",
i === 0 ? "good-news" : "bad-news",
),
});
if (this.isFifo) {
queueName += ".fifo";
await this.logger.log(MESSAGES.appendFifoNotice);
}
const response = await this.sqsClient.send(
new CreateQueueCommand({
QueueName: queueName,
Attributes: { ...(this.isFifo ? { FifoQueue: "true" } : {}) },
}),
);
const { Attributes } = await this.sqsClient.send(
new GetQueueAttributesCommand({
QueueUrl: response.QueueUrl,
AttributeNames: ["QueueArn"],
}),
);
this.queues.push({
queueName,
queueArn: Attributes.QueueArn,
queueUrl: response.QueueUrl,
});
await this.logger.log(
MESSAGES.queueCreatedNotice
.replace("${QUEUE_NAME}", queueName)
.replace("${QUEUE_URL}", response.QueueUrl)
.replace("${QUEUE_ARN}", Attributes.QueueArn),
);
}
}
async attachQueueIamPolicies() {
for (const [index, queue] of this.queues.entries()) {
const policy = JSON.stringify(
{
Statement: [
{
Effect: "Allow",
Principal: {
Service: "sns.amazonaws.com",
},
Action: "sqs:SendMessage",
Resource: queue.queueArn,
Condition: {
ArnEquals: {
"aws:SourceArn": this.topicArn,
},
},
},
],
},
null,
2,
);
if (index !== 0) {
this.logger.logSeparator();
}
await this.logger.log(MESSAGES.attachPolicyNotice);
console.log(policy);
const addPolicy = await this.prompter.confirm({
message: MESSAGES.addPolicyConfirmation.replace(
"${QUEUE_NAME}",
queue.queueName,
),
});
if (addPolicy) {
await this.sqsClient.send(
new SetQueueAttributesCommand({
QueueUrl: queue.queueUrl,
Attributes: {
Policy: policy,
},
}),
);
queue.policy = policy;
} else {
await this.logger.log(
MESSAGES.policyNotAttachedNotice.replace(
"${QUEUE_NAME}",
queue.queueName,
),
);
}
}
}
async subscribeQueuesToTopic() {
for (const [index, queue] of this.queues.entries()) {
/**
* @type {import('@aws-sdk/client-sns').SubscribeCommandInput}
*/
const subscribeParams = {
TopicArn: this.topicArn,
Protocol: "sqs",
Endpoint: queue.queueArn,
};
let tones = [];
if (this.isFifo) {
if (index === 0) {
await this.logger.log(MESSAGES.fifoFilterNotice);
}
tones = await this.prompter.checkbox({
message: MESSAGES.fifoFilterSelect.replace(
"${QUEUE_NAME}",
queue.queueName,
),
choices: toneChoices,
});
if (tones.length) {
subscribeParams.Attributes = {
FilterPolicyScope: "MessageAttributes",
FilterPolicy: JSON.stringify({
tone: tones,
}),
};
}
}
const { SubscriptionArn } = await this.snsClient.send(
new SubscribeCommand(subscribeParams),
);
this.subscriptionArns.push(SubscriptionArn);
await this.logger.log(
MESSAGES.queueSubscribedNotice
.replace("${QUEUE_NAME}", queue.queueName)
.replace("${TOPIC_NAME}", this.topicName)
.replace("${TONES}", tones.length ? tones.join(", ") : "none"),
);
}
}
async publishMessages() {
const message = await this.prompter.input({
message: MESSAGES.publishMessagePrompt,
});
let groupId;
let deduplicationId;
let choices;
if (this.isFifo) {
await this.logger.log(MESSAGES.groupIdNotice);
groupId = await this.prompter.input({
message: MESSAGES.groupIdPrompt,
});
if (this.autoDedup === false) {
await this.logger.log(MESSAGES.deduplicationIdNotice);
deduplicationId = await this.prompter.input({
message: MESSAGES.deduplicationIdPrompt,
});
}
choices = await this.prompter.checkbox({
message: MESSAGES.messageAttributesPrompt,
choices: toneChoices,
});
}
await this.snsClient.send(
new PublishCommand({
TopicArn: this.topicArn,
Message: message,
...(groupId
? {
MessageGroupId: groupId,
}
: {}),
...(deduplicationId
? {
MessageDeduplicationId: deduplicationId,
}
: {}),
...(choices
? {
MessageAttributes: {
tone: {
DataType: "String.Array",
StringValue: JSON.stringify(choices),
},
},
}
: {}),
}),
);
const publishAnother = await this.prompter.confirm({
message: MESSAGES.publishAnother,
});
if (publishAnother) {
await this.publishMessages();
}
}
async receiveAndDeleteMessages() {
for (const queue of this.queues) {
const { Messages } = await this.sqsClient.send(
new ReceiveMessageCommand({
QueueUrl: queue.queueUrl,
}),
);
if (Messages) {
await this.logger.log(
MESSAGES.messagesReceivedNotice.replace(
"${QUEUE_NAME}",
queue.queueName,
),
);
console.log(Messages);
await this.sqsClient.send(
new DeleteMessageBatchCommand({
QueueUrl: queue.queueUrl,
Entries: Messages.map((message) => ({
Id: message.MessageId,
ReceiptHandle: message.ReceiptHandle,
})),
}),
);
} else {
await this.logger.log(
MESSAGES.noMessagesReceivedNotice.replace(
"${QUEUE_NAME}",
queue.queueName,
),
);
}
}
const deleteAndPoll = await this.prompter.confirm({
message: MESSAGES.deleteAndPollConfirmation,
});
if (deleteAndPoll) {
await this.receiveAndDeleteMessages();
}
}
async destroyResources() {
for (const subscriptionArn of this.subscriptionArns) {
await this.snsClient.send(
new UnsubscribeCommand({ SubscriptionArn: subscriptionArn }),
);
}
for (const queue of this.queues) {
await this.sqsClient.send(
new DeleteQueueCommand({ QueueUrl: queue.queueUrl }),
);
}
if (this.topicArn) {
await this.snsClient.send(
new DeleteTopicCommand({ TopicArn: this.topicArn }),
);
}
}
async start() {
console.clear();
try {
this.logger.logSeparator(MESSAGES.headerWelcome);
await this.welcome();
this.logger.logSeparator(MESSAGES.headerFifo);
await this.confirmFifo();
this.logger.logSeparator(MESSAGES.headerCreateTopic);
await this.createTopic();
this.logger.logSeparator(MESSAGES.headerCreateQueues);
await this.createQueues();
this.logger.logSeparator(MESSAGES.headerAttachPolicy);
await this.attachQueueIamPolicies();
this.logger.logSeparator(MESSAGES.headerSubscribeQueues);
await this.subscribeQueuesToTopic();
this.logger.logSeparator(MESSAGES.headerPublishMessage);
await this.publishMessages();
this.logger.logSeparator(MESSAGES.headerReceiveMessages);
await this.receiveAndDeleteMessages();
} catch (err) {
console.error(err);
} finally {
await this.destroyResources();
}
}
}
```
+ For API details, see the following topics in *AWS SDK for JavaScript API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs/command/CreateQueueCommand)
+ [CreateTopic](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns/command/CreateTopicCommand)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs/command/DeleteMessageBatchCommand)
+ [DeleteQueue](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs/command/DeleteQueueCommand)
+ [DeleteTopic](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns/command/DeleteTopicCommand)
+ [GetQueueAttributes](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs/command/GetQueueAttributesCommand)
+ [Publish](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns/command/PublishCommand)
+ [ReceiveMessage](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs/command/ReceiveMessageCommand)
+ [SetQueueAttributes](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sqs/command/SetQueueAttributesCommand)
+ [Subscribe](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns/command/SubscribeCommand)
+ [Unsubscribe](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/sns/command/UnsubscribeCommand)
------
#### [ Kotlin ]
**SDK for Kotlin**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/kotlin/usecases/topics_and_queues#code-examples).
```
package com.example.sns
import aws.sdk.kotlin.services.sns.SnsClient
import aws.sdk.kotlin.services.sns.model.CreateTopicRequest
import aws.sdk.kotlin.services.sns.model.DeleteTopicRequest
import aws.sdk.kotlin.services.sns.model.PublishRequest
import aws.sdk.kotlin.services.sns.model.SetSubscriptionAttributesRequest
import aws.sdk.kotlin.services.sns.model.SubscribeRequest
import aws.sdk.kotlin.services.sns.model.UnsubscribeRequest
import aws.sdk.kotlin.services.sqs.SqsClient
import aws.sdk.kotlin.services.sqs.model.CreateQueueRequest
import aws.sdk.kotlin.services.sqs.model.DeleteMessageBatchRequest
import aws.sdk.kotlin.services.sqs.model.DeleteMessageBatchRequestEntry
import aws.sdk.kotlin.services.sqs.model.DeleteQueueRequest
import aws.sdk.kotlin.services.sqs.model.GetQueueAttributesRequest
import aws.sdk.kotlin.services.sqs.model.GetQueueUrlRequest
import aws.sdk.kotlin.services.sqs.model.Message
import aws.sdk.kotlin.services.sqs.model.QueueAttributeName
import aws.sdk.kotlin.services.sqs.model.ReceiveMessageRequest
import aws.sdk.kotlin.services.sqs.model.SetQueueAttributesRequest
import com.google.gson.Gson
import com.google.gson.JsonObject
import com.google.gson.JsonPrimitive
import java.util.Scanner
/**
Before running this Kotlin code example, set up your development environment,
including your AWS credentials.
For more information, see the following documentation topic:
https://docs.aws.amazon.com/sdk-for-kotlin/latest/developer-guide/setup.html
This Kotlin example performs the following tasks:
1. Gives the user three options to choose from.
2. Creates an Amazon Simple Notification Service (Amazon SNS) topic.
3. Creates an Amazon Simple Queue Service (Amazon SQS) queue.
4. Gets the SQS queue Amazon Resource Name (ARN) attribute.
5. Attaches an AWS Identity and Access Management (IAM) policy to the queue.
6. Subscribes to the SQS queue.
7. Publishes a message to the topic.
8. Displays the messages.
9. Deletes the received message.
10. Unsubscribes from the topic.
11. Deletes the SNS topic.
*/
val DASHES: String = String(CharArray(80)).replace("\u0000", "-")
suspend fun main() {
val input = Scanner(System.`in`)
val useFIFO: String
var duplication = "n"
var topicName: String
var deduplicationID: String? = null
var groupId: String? = null
val topicArn: String?
var sqsQueueName: String
val sqsQueueUrl: String?
val sqsQueueArn: String
val subscriptionArn: String?
var selectFIFO = false
val message: String
val messageList: List?
val filterList = ArrayList()
var msgAttValue = ""
println(DASHES)
println("Welcome to the AWS SDK for Kotlin messaging with topics and queues.")
println(
"""
In this scenario, you will create an SNS topic and subscribe an SQS queue to the topic.
You can select from several options for configuring the topic and the subscriptions for the queue.
You can then post to the topic and see the results in the queue.
""".trimIndent(),
)
println(DASHES)
println(DASHES)
println(
"""
SNS topics can be configured as FIFO (First-In-First-Out).
FIFO topics deliver messages in order and support deduplication and message filtering.
Would you like to work with FIFO topics? (y/n)
""".trimIndent(),
)
useFIFO = input.nextLine()
if (useFIFO.compareTo("y") == 0) {
selectFIFO = true
println("You have selected FIFO")
println(
""" Because you have chosen a FIFO topic, deduplication is supported.
Deduplication IDs are either set in the message or automatically generated from content using a hash function.
If a message is successfully published to an SNS FIFO topic, any message published and determined to have the same deduplication ID,
within the five-minute deduplication interval, is accepted but not delivered.
For more information about deduplication, see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html.""",
)
println("Would you like to use content-based deduplication instead of entering a deduplication ID? (y/n)")
duplication = input.nextLine()
if (duplication.compareTo("y") == 0) {
println("Enter a group id value")
groupId = input.nextLine()
} else {
println("Enter deduplication Id value")
deduplicationID = input.nextLine()
println("Enter a group id value")
groupId = input.nextLine()
}
}
println(DASHES)
println(DASHES)
println("2. Create a topic.")
println("Enter a name for your SNS topic.")
topicName = input.nextLine()
if (selectFIFO) {
println("Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.")
topicName = "$topicName.fifo"
println("The name of the topic is $topicName")
topicArn = createFIFO(topicName, duplication)
println("The ARN of the FIFO topic is $topicArn")
} else {
println("The name of the topic is $topicName")
topicArn = createSNSTopic(topicName)
println("The ARN of the non-FIFO topic is $topicArn")
}
println(DASHES)
println(DASHES)
println("3. Create an SQS queue.")
println("Enter a name for your SQS queue.")
sqsQueueName = input.nextLine()
if (selectFIFO) {
sqsQueueName = "$sqsQueueName.fifo"
}
sqsQueueUrl = createQueue(sqsQueueName, selectFIFO)
println("The queue URL is $sqsQueueUrl")
println(DASHES)
println(DASHES)
println("4. Get the SQS queue ARN attribute.")
sqsQueueArn = getSQSQueueAttrs(sqsQueueUrl)
println("The ARN of the new queue is $sqsQueueArn")
println(DASHES)
println(DASHES)
println("5. Attach an IAM policy to the queue.")
// Define the policy to use.
val policy = """{
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "sns.amazonaws.com"
},
"Action": "sqs:SendMessage",
"Resource": "$sqsQueueArn",
"Condition": {
"ArnEquals": {
"aws:SourceArn": "$topicArn"
}
}
}
]
}"""
setQueueAttr(sqsQueueUrl, policy)
println(DASHES)
println(DASHES)
println("6. Subscribe to the SQS queue.")
if (selectFIFO) {
println(
"""If you add a filter to this subscription, then only the filtered messages will be received in the queue.
For information about message filtering, see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html
For this example, you can filter messages by a "tone" attribute.""",
)
println("Would you like to filter messages for $sqsQueueName's subscription to the topic $topicName? (y/n)")
val filterAns: String = input.nextLine()
if (filterAns.compareTo("y") == 0) {
var moreAns = false
println("You can filter messages by using one or more of the following \"tone\" attributes.")
println("1. cheerful")
println("2. funny")
println("3. serious")
println("4. sincere")
while (!moreAns) {
println("Select a number or choose 0 to end.")
val ans: String = input.nextLine()
when (ans) {
"1" -> filterList.add("cheerful")
"2" -> filterList.add("funny")
"3" -> filterList.add("serious")
"4" -> filterList.add("sincere")
else -> moreAns = true
}
}
}
}
subscriptionArn = subQueue(topicArn, sqsQueueArn, filterList)
println(DASHES)
println(DASHES)
println("7. Publish a message to the topic.")
if (selectFIFO) {
println("Would you like to add an attribute to this message? (y/n)")
val msgAns: String = input.nextLine()
if (msgAns.compareTo("y") == 0) {
println("You can filter messages by one or more of the following \"tone\" attributes.")
println("1. cheerful")
println("2. funny")
println("3. serious")
println("4. sincere")
println("Select a number or choose 0 to end.")
val ans: String = input.nextLine()
msgAttValue = when (ans) {
"1" -> "cheerful"
"2" -> "funny"
"3" -> "serious"
else -> "sincere"
}
println("Selected value is $msgAttValue")
}
println("Enter a message.")
message = input.nextLine()
pubMessageFIFO(message, topicArn, msgAttValue, duplication, groupId, deduplicationID)
} else {
println("Enter a message.")
message = input.nextLine()
pubMessage(message, topicArn)
}
println(DASHES)
println(DASHES)
println("8. Display the message. Press any key to continue.")
input.nextLine()
messageList = receiveMessages(sqsQueueUrl, msgAttValue)
if (messageList != null) {
for (mes in messageList) {
println("Message Id: ${mes.messageId}")
println("Full Message: ${mes.body}")
}
}
println(DASHES)
println(DASHES)
println("9. Delete the received message. Press any key to continue.")
input.nextLine()
if (messageList != null) {
deleteMessages(sqsQueueUrl, messageList)
}
println(DASHES)
println(DASHES)
println("10. Unsubscribe from the topic and delete the queue. Press any key to continue.")
input.nextLine()
unSub(subscriptionArn)
deleteSQSQueue(sqsQueueName)
println(DASHES)
println(DASHES)
println("11. Delete the topic. Press any key to continue.")
input.nextLine()
deleteSNSTopic(topicArn)
println(DASHES)
println(DASHES)
println("The SNS/SQS workflow has completed successfully.")
println(DASHES)
}
suspend fun deleteSNSTopic(topicArnVal: String?) {
val request = DeleteTopicRequest {
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
snsClient.deleteTopic(request)
println("$topicArnVal was deleted")
}
}
suspend fun deleteSQSQueue(queueNameVal: String) {
val getQueueRequest = GetQueueUrlRequest {
queueName = queueNameVal
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
val queueUrlVal = sqsClient.getQueueUrl(getQueueRequest).queueUrl
val deleteQueueRequest = DeleteQueueRequest {
queueUrl = queueUrlVal
}
sqsClient.deleteQueue(deleteQueueRequest)
println("$queueNameVal was successfully deleted.")
}
}
suspend fun unSub(subscripArn: String?) {
val request = UnsubscribeRequest {
subscriptionArn = subscripArn
}
SnsClient { region = "us-east-1" }.use { snsClient ->
snsClient.unsubscribe(request)
println("Subscription was removed for $subscripArn")
}
}
suspend fun deleteMessages(queueUrlVal: String?, messages: List) {
val entriesVal: MutableList = mutableListOf()
for (msg in messages) {
val entry = DeleteMessageBatchRequestEntry {
id = msg.messageId
}
entriesVal.add(entry)
}
val deleteMessageBatchRequest = DeleteMessageBatchRequest {
queueUrl = queueUrlVal
entries = entriesVal
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
sqsClient.deleteMessageBatch(deleteMessageBatchRequest)
println("The batch delete of messages was successful")
}
}
suspend fun receiveMessages(queueUrlVal: String?, msgAttValue: String): List? {
if (msgAttValue.isEmpty()) {
val request = ReceiveMessageRequest {
queueUrl = queueUrlVal
maxNumberOfMessages = 5
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
return sqsClient.receiveMessage(request).messages
}
} else {
val receiveRequest = ReceiveMessageRequest {
queueUrl = queueUrlVal
waitTimeSeconds = 1
maxNumberOfMessages = 5
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
return sqsClient.receiveMessage(receiveRequest).messages
}
}
}
suspend fun pubMessage(messageVal: String?, topicArnVal: String?) {
val request = PublishRequest {
message = messageVal
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.publish(request)
println("${result.messageId} message sent.")
}
}
suspend fun pubMessageFIFO(
messageVal: String?,
topicArnVal: String?,
msgAttValue: String,
duplication: String,
groupIdVal: String?,
deduplicationID: String?,
) {
// Means the user did not choose to use a message attribute.
if (msgAttValue.isEmpty()) {
if (duplication.compareTo("y") == 0) {
val request = PublishRequest {
message = messageVal
messageGroupId = groupIdVal
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.publish(request)
println(result.messageId.toString() + " Message sent.")
}
} else {
val request = PublishRequest {
message = messageVal
messageDeduplicationId = deduplicationID
messageGroupId = groupIdVal
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.publish(request)
println(result.messageId.toString() + " Message sent.")
}
}
} else {
val messAttr = aws.sdk.kotlin.services.sns.model.MessageAttributeValue {
dataType = "String"
stringValue = "true"
}
val mapAtt: Map =
mapOf(msgAttValue to messAttr)
if (duplication.compareTo("y") == 0) {
val request = PublishRequest {
message = messageVal
messageGroupId = groupIdVal
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.publish(request)
println(result.messageId.toString() + " Message sent.")
}
} else {
// Create a publish request with the message and attributes.
val request = PublishRequest {
topicArn = topicArnVal
message = messageVal
messageDeduplicationId = deduplicationID
messageGroupId = groupIdVal
messageAttributes = mapAtt
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.publish(request)
println(result.messageId.toString() + " Message sent.")
}
}
}
}
// Subscribe to the SQS queue.
suspend fun subQueue(topicArnVal: String?, queueArnVal: String, filterList: List): String? {
val request: SubscribeRequest
if (filterList.isEmpty()) {
// No filter subscription is added.
request = SubscribeRequest {
protocol = "sqs"
endpoint = queueArnVal
returnSubscriptionArn = true
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.subscribe(request)
println(
"The queue " + queueArnVal + " has been subscribed to the topic " + topicArnVal + "\n" +
"with the subscription ARN " + result.subscriptionArn,
)
return result.subscriptionArn
}
} else {
request = SubscribeRequest {
protocol = "sqs"
endpoint = queueArnVal
returnSubscriptionArn = true
topicArn = topicArnVal
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.subscribe(request)
println("The queue $queueArnVal has been subscribed to the topic $topicArnVal with the subscription ARN ${result.subscriptionArn}")
val attributeNameVal = "FilterPolicy"
val gson = Gson()
val jsonString = "{\"tone\": []}"
val jsonObject = gson.fromJson(jsonString, JsonObject::class.java)
val toneArray = jsonObject.getAsJsonArray("tone")
for (value: String? in filterList) {
toneArray.add(JsonPrimitive(value))
}
val updatedJsonString: String = gson.toJson(jsonObject)
println(updatedJsonString)
val attRequest = SetSubscriptionAttributesRequest {
subscriptionArn = result.subscriptionArn
attributeName = attributeNameVal
attributeValue = updatedJsonString
}
snsClient.setSubscriptionAttributes(attRequest)
return result.subscriptionArn
}
}
}
suspend fun setQueueAttr(queueUrlVal: String?, policy: String) {
val attrMap: MutableMap = HashMap()
attrMap[QueueAttributeName.Policy.toString()] = policy
val attributesRequest = SetQueueAttributesRequest {
queueUrl = queueUrlVal
attributes = attrMap
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
sqsClient.setQueueAttributes(attributesRequest)
println("The policy has been successfully attached.")
}
}
suspend fun getSQSQueueAttrs(queueUrlVal: String?): String {
val atts: MutableList = ArrayList()
atts.add(QueueAttributeName.QueueArn)
val attributesRequest = GetQueueAttributesRequest {
queueUrl = queueUrlVal
attributeNames = atts
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
val response = sqsClient.getQueueAttributes(attributesRequest)
val mapAtts = response.attributes
if (mapAtts != null) {
mapAtts.forEach { entry ->
println("${entry.key} : ${entry.value}")
return entry.value
}
}
}
return ""
}
suspend fun createQueue(queueNameVal: String?, selectFIFO: Boolean): String? {
println("\nCreate Queue")
if (selectFIFO) {
val attrs = mutableMapOf()
attrs[QueueAttributeName.FifoQueue.toString()] = "true"
val createQueueRequest = CreateQueueRequest {
queueName = queueNameVal
attributes = attrs
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
sqsClient.createQueue(createQueueRequest)
println("\nGet queue url")
val urlRequest = GetQueueUrlRequest {
queueName = queueNameVal
}
val getQueueUrlResponse = sqsClient.getQueueUrl(urlRequest)
return getQueueUrlResponse.queueUrl
}
} else {
val createQueueRequest = CreateQueueRequest {
queueName = queueNameVal
}
SqsClient { region = "us-east-1" }.use { sqsClient ->
sqsClient.createQueue(createQueueRequest)
println("Get queue url")
val urlRequest = GetQueueUrlRequest {
queueName = queueNameVal
}
val getQueueUrlResponse = sqsClient.getQueueUrl(urlRequest)
return getQueueUrlResponse.queueUrl
}
}
}
suspend fun createSNSTopic(topicName: String?): String? {
val request = CreateTopicRequest {
name = topicName
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val result = snsClient.createTopic(request)
return result.topicArn
}
}
suspend fun createFIFO(topicName: String?, duplication: String): String? {
val topicAttributes: MutableMap = HashMap()
if (duplication.compareTo("n") == 0) {
topicAttributes["FifoTopic"] = "true"
topicAttributes["ContentBasedDeduplication"] = "false"
} else {
topicAttributes["FifoTopic"] = "true"
topicAttributes["ContentBasedDeduplication"] = "true"
}
val topicRequest = CreateTopicRequest {
name = topicName
attributes = topicAttributes
}
SnsClient { region = "us-east-1" }.use { snsClient ->
val response = snsClient.createTopic(topicRequest)
return response.topicArn
}
}
```
+ For API details, see the following topics in *AWS SDK for Kotlin API reference*.
+ [CreateQueue](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [CreateTopic](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [DeleteMessageBatch](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [DeleteQueue](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [DeleteTopic](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [GetQueueAttributes](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [Publish](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [ReceiveMessage](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [SetQueueAttributes](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [Subscribe](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
+ [Unsubscribe](https://sdk.amazonaws.com/kotlin/api/latest/index.html)
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/topics_and_queues#code-examples).
Run an interactive scenario at a command prompt.
```
class TopicsAndQueuesScenario:
"""Manages the Topics and Queues feature scenario."""
DASHES = "-" * 80
def __init__(self, sns_wrapper: SnsWrapper, sqs_wrapper: SqsWrapper) -> None:
"""
Initialize the Topics and Queues scenario.
:param sns_wrapper: SnsWrapper instance for SNS operations.
:param sqs_wrapper: SqsWrapper instance for SQS operations.
"""
self.sns_wrapper = sns_wrapper
self.sqs_wrapper = sqs_wrapper
# Scenario state
self.use_fifo_topic = False
self.use_content_based_deduplication = False
self.topic_name = None
self.topic_arn = None
self.queue_count = 2
self.queue_urls = []
self.subscription_arns = []
self.tones = ["cheerful", "funny", "serious", "sincere"]
def run_scenario(self) -> None:
"""Run the Topics and Queues feature scenario."""
print(self.DASHES)
print("Welcome to messaging with topics and queues.")
print(self.DASHES)
print(f"""
In this scenario, you will create an SNS topic and subscribe {self.queue_count} SQS queues to the topic.
You can select from several options for configuring the topic and the subscriptions for the queues.
You can then post to the topic and see the results in the queues.
""")
try:
# Setup Phase
print(self.DASHES)
self._setup_topic()
print(self.DASHES)
self._setup_queues()
print(self.DASHES)
# Demonstration Phase
self._publish_messages()
print(self.DASHES)
# Examination Phase
self._poll_queues_for_messages()
print(self.DASHES)
# Cleanup Phase
self._cleanup_resources()
print(self.DASHES)
except Exception as e:
logger.error(f"Scenario failed: {e}")
print(f"There was a problem with the scenario: {e}")
print("\nInitiating cleanup...")
try:
self._cleanup_resources()
except Exception as cleanup_error:
logger.error(f"Error during cleanup: {cleanup_error}")
print("Messaging with topics and queues scenario is complete.")
print(self.DASHES)
def _setup_topic(self) -> None:
"""Set up the SNS topic to be used with the queues."""
print("SNS topics can be configured as FIFO (First-In-First-Out).")
print("FIFO topics deliver messages in order and support deduplication and message filtering.")
print()
self.use_fifo_topic = q.ask("Would you like to work with FIFO topics? (y/n): ", q.is_yesno)
if self.use_fifo_topic:
print(self.DASHES)
self.topic_name = q.ask("Enter a name for your SNS topic: ", q.non_empty)
print("Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.")
print()
print(self.DASHES)
print("""
Because you have chosen a FIFO topic, deduplication is supported.
Deduplication IDs are either set in the message or automatically generated
from content using a hash function.
If a message is successfully published to an SNS FIFO topic, any message
published and determined to have the same deduplication ID,
within the five-minute deduplication interval, is accepted but not delivered.
For more information about deduplication,
see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html.
""")
self.use_content_based_deduplication = q.ask(
"Use content-based deduplication instead of entering a deduplication ID? (y/n): ",
q.is_yesno
)
else:
self.topic_name = q.ask("Enter a name for your SNS topic: ", q.non_empty)
print(self.DASHES)
# Create the topic
self.topic_arn = self.sns_wrapper.create_topic(
self.topic_name,
self.use_fifo_topic,
self.use_content_based_deduplication
)
print(f"Your new topic with the name {self.topic_name}")
print(f" and Amazon Resource Name (ARN) {self.topic_arn}")
print(f" has been created.")
print()
def _setup_queues(self) -> None:
"""Set up the SQS queues and subscribe them to the topic."""
print(f"Now you will create {self.queue_count} Amazon Simple Queue Service (Amazon SQS) queues to subscribe to the topic.")
for i in range(self.queue_count):
queue_name = q.ask(f"Enter a name for SQS queue #{i+1}: ", q.non_empty)
if self.use_fifo_topic and i == 0:
print("Because you have selected a FIFO topic, '.fifo' must be appended to the queue name.")
# Create the queue
queue_url = self.sqs_wrapper.create_queue(queue_name, self.use_fifo_topic)
self.queue_urls.append(queue_url)
print(f"Your new queue with the name {queue_name}")
print(f" and queue URL {queue_url}")
print(f" has been created.")
print()
if i == 0:
print("The queue URL is used to retrieve the queue ARN,")
print("which is used to create a subscription.")
print(self.DASHES)
# Get queue ARN
queue_arn = self.sqs_wrapper.get_queue_arn(queue_url)
if i == 0:
print("An AWS Identity and Access Management (IAM) policy must be attached to an SQS queue,")
print("enabling it to receive messages from an SNS topic.")
# Set queue policy to allow SNS to send messages
self.sqs_wrapper.set_queue_policy_for_topic(queue_arn, self.topic_arn, queue_url)
# Set up message filtering if using FIFO
subscription_arn = self._setup_subscription_with_filter(i, queue_arn, queue_name)
self.subscription_arns.append(subscription_arn)
def _setup_subscription_with_filter(self, queue_index: int, queue_arn: str, queue_name: str) -> str:
"""Set up subscription with optional message filtering."""
filter_policy = None
if self.use_fifo_topic:
print(self.DASHES)
if queue_index == 0:
print("Subscriptions to a FIFO topic can have filters.")
print("If you add a filter to this subscription, then only the filtered messages")
print("will be received in the queue.")
print()
print("For information about message filtering,")
print("see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html")
print()
print("For this example, you can filter messages by a TONE attribute.")
use_filter = q.ask(f"Filter messages for {queue_name}'s subscription to the topic? (y/n): ", q.is_yesno)
if use_filter:
filter_policy = self._create_filter_policy()
subscription_arn = self.sns_wrapper.subscribe_queue_to_topic(
self.topic_arn, queue_arn, filter_policy
)
print(f"The queue {queue_name} has been subscribed to the topic {self.topic_name}")
print(f" with the subscription ARN {subscription_arn}")
return subscription_arn
def _create_filter_policy(self) -> str:
"""Create a message filter policy based on user selections."""
print(self.DASHES)
print("You can filter messages by one or more of the following TONE attributes.")
filter_selections = []
selection_number = 0
while True:
print("Enter a number to add a TONE filter, or enter 0 to stop adding filters.")
for i, tone in enumerate(self.tones, 1):
print(f" {i}. {tone}")
selection = q.ask("Your choice: ", q.is_int, q.in_range(0, len(self.tones)))
if selection == 0:
break
elif selection > 0 and self.tones[selection - 1] not in filter_selections:
filter_selections.append(self.tones[selection - 1])
print(f"Added '{self.tones[selection - 1]}' to filter list.")
if filter_selections:
filters = {"tone": filter_selections}
return json.dumps(filters)
return None
def _publish_messages(self) -> None:
"""Publish messages to the topic with various options."""
print("Now we can publish messages.")
keep_sending = True
while keep_sending:
print()
message = q.ask("Enter a message to publish: ", q.non_empty)
message_group_id = None
deduplication_id = None
tone_attribute = None
if self.use_fifo_topic:
print("Because you are using a FIFO topic, you must set a message group ID.")
print("All messages within the same group will be received in the order they were published.")
print()
message_group_id = q.ask("Enter a message group ID for this message: ", q.non_empty)
if not self.use_content_based_deduplication:
print("Because you are not using content-based deduplication,")
print("you must enter a deduplication ID.")
deduplication_id = q.ask("Enter a deduplication ID for this message: ", q.non_empty)
# Ask about tone attribute
add_attribute = q.ask("Add an attribute to this message? (y/n): ", q.is_yesno)
if add_attribute:
print("Enter a number for an attribute:")
for i, tone in enumerate(self.tones, 1):
print(f" {i}. {tone}")
selection = q.ask("Your choice: ", q.is_int, q.in_range(1, len(self.tones)))
if 1 <= selection <= len(self.tones):
tone_attribute = self.tones[selection - 1]
# Publish the message
message_id = self.sns_wrapper.publish_message(
self.topic_arn,
message,
tone_attribute,
deduplication_id,
message_group_id
)
print(f"Message published with ID: {message_id}")
keep_sending = q.ask("Send another message? (y/n): ", q.is_yesno)
def _poll_queues_for_messages(self) -> None:
"""Poll all queues for messages and display results."""
for i, queue_url in enumerate(self.queue_urls):
print(f"Polling queue #{i+1} at {queue_url} for messages...")
q.ask("Press Enter to continue...")
messages = self._poll_queue_for_messages(queue_url)
if messages:
print(f"{len(messages)} message(s) were received by queue #{i+1}")
for j, message in enumerate(messages, 1):
print(f" Message {j}:")
# Parse the SNS message body to get the actual message
try:
sns_message = json.loads(message['Body'])
actual_message = sns_message.get('Message', message['Body'])
print(f" {actual_message}")
except (json.JSONDecodeError, KeyError):
print(f" {message['Body']}")
# Delete the messages
self.sqs_wrapper.delete_messages(queue_url, messages)
print(f"Messages deleted from queue #{i+1}")
else:
print(f"No messages received by queue #{i+1}")
print(self.DASHES)
def _poll_queue_for_messages(self, queue_url: str) -> List[Dict[str, Any]]:
"""Poll a single queue for messages."""
all_messages = []
max_polls = 3 # Limit polling to avoid infinite loops
for poll_count in range(max_polls):
messages = self.sqs_wrapper.receive_messages(queue_url, 10)
if messages:
all_messages.extend(messages)
print(f" Received {len(messages)} messages in poll {poll_count + 1}")
# Small delay between polls
time.sleep(1)
else:
print(f" No messages in poll {poll_count + 1}")
break
return all_messages
def _cleanup_resources(self) -> None:
"""Clean up all resources created during the scenario."""
print("Cleaning up resources...")
# Delete queues
for i, queue_url in enumerate(self.queue_urls):
if queue_url:
delete_queue = q.ask(f"Delete queue #{i+1} with URL {queue_url}? (y/n): ", q.is_yesno)
if delete_queue:
try:
self.sqs_wrapper.delete_queue(queue_url)
print(f"Deleted queue #{i+1}")
except Exception as e:
print(f"Error deleting queue #{i+1}: {e}")
# Unsubscribe from topic
for i, subscription_arn in enumerate(self.subscription_arns):
if subscription_arn:
try:
self.sns_wrapper.unsubscribe(subscription_arn)
print(f"Unsubscribed subscription #{i+1}")
except Exception as e:
print(f"Error unsubscribing #{i+1}: {e}")
# Delete topic
if self.topic_arn:
delete_topic = q.ask(f"Delete topic {self.topic_name}? (y/n): ", q.is_yesno)
if delete_topic:
try:
self.sns_wrapper.delete_topic(self.topic_arn)
print(f"Deleted topic {self.topic_name}")
except Exception as e:
print(f"Error deleting topic: {e}")
print("Resource cleanup complete.")
```
Create classes that wrap Amazon SNS and Amazon SQS operations for use in the scenario.
```
class SnsWrapper:
"""Wrapper class for managing Amazon SNS operations."""
def __init__(self, sns_client: Any) -> None:
"""
Initialize the SnsWrapper.
:param sns_client: A Boto3 Amazon SNS client.
"""
self.sns_client = sns_client
@classmethod
def from_client(cls) -> 'SnsWrapper':
"""
Create an SnsWrapper instance using a default boto3 client.
:return: An instance of this class.
"""
sns_client = boto3.client('sns')
return cls(sns_client)
def create_topic(
self,
topic_name: str,
is_fifo: bool = False,
content_based_deduplication: bool = False
) -> str:
"""
Create an SNS topic.
:param topic_name: The name of the topic to create.
:param is_fifo: Whether to create a FIFO topic.
:param content_based_deduplication: Whether to use content-based deduplication for FIFO topics.
:return: The ARN of the created topic.
:raises ClientError: If the topic creation fails.
"""
try:
# Add .fifo suffix for FIFO topics
if is_fifo and not topic_name.endswith('.fifo'):
topic_name += '.fifo'
attributes = {}
if is_fifo:
attributes['FifoTopic'] = 'true'
if content_based_deduplication:
attributes['ContentBasedDeduplication'] = 'true'
response = self.sns_client.create_topic(
Name=topic_name,
Attributes=attributes
)
topic_arn = response['TopicArn']
logger.info(f"Created topic: {topic_name} with ARN: {topic_arn}")
return topic_arn
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error creating topic {topic_name}: {error_code} - {e}")
raise
def subscribe_queue_to_topic(
self,
topic_arn: str,
queue_arn: str,
filter_policy: Optional[str] = None
) -> str:
"""
Subscribe an SQS queue to an SNS topic.
:param topic_arn: The ARN of the SNS topic.
:param queue_arn: The ARN of the SQS queue.
:param filter_policy: Optional JSON filter policy for message filtering.
:return: The ARN of the subscription.
:raises ClientError: If the subscription fails.
"""
try:
attributes = {}
if filter_policy:
attributes['FilterPolicy'] = filter_policy
response = self.sns_client.subscribe(
TopicArn=topic_arn,
Protocol='sqs',
Endpoint=queue_arn,
Attributes=attributes
)
subscription_arn = response['SubscriptionArn']
logger.info(f"Subscribed queue {queue_arn} to topic {topic_arn}")
return subscription_arn
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error subscribing queue to topic: {error_code} - {e}")
raise
def publish_message(
self,
topic_arn: str,
message: str,
tone_attribute: Optional[str] = None,
deduplication_id: Optional[str] = None,
message_group_id: Optional[str] = None
) -> str:
"""
Publish a message to an SNS topic.
:param topic_arn: The ARN of the SNS topic.
:param message: The message content to publish.
:param tone_attribute: Optional tone attribute for message filtering.
:param deduplication_id: Optional deduplication ID for FIFO topics.
:param message_group_id: Optional message group ID for FIFO topics.
:return: The message ID of the published message.
:raises ClientError: If the message publication fails.
"""
try:
publish_args = {
'TopicArn': topic_arn,
'Message': message
}
# Add message attributes if tone is specified
if tone_attribute:
publish_args['MessageAttributes'] = {
'tone': {
'DataType': 'String',
'StringValue': tone_attribute
}
}
# Add FIFO-specific parameters
if message_group_id:
publish_args['MessageGroupId'] = message_group_id
if deduplication_id:
publish_args['MessageDeduplicationId'] = deduplication_id
response = self.sns_client.publish(**publish_args)
message_id = response['MessageId']
logger.info(f"Published message to topic {topic_arn} with ID: {message_id}")
return message_id
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error publishing message to topic: {error_code} - {e}")
raise
def unsubscribe(self, subscription_arn: str) -> bool:
"""
Unsubscribe from an SNS topic.
:param subscription_arn: The ARN of the subscription to remove.
:return: True if successful.
:raises ClientError: If the unsubscribe operation fails.
"""
try:
self.sns_client.unsubscribe(SubscriptionArn=subscription_arn)
logger.info(f"Unsubscribed: {subscription_arn}")
return True
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
if error_code == 'NotFound':
logger.warning(f"Subscription not found: {subscription_arn}")
return True # Already unsubscribed
else:
logger.error(f"Error unsubscribing: {error_code} - {e}")
raise
def delete_topic(self, topic_arn: str) -> bool:
"""
Delete an SNS topic.
:param topic_arn: The ARN of the topic to delete.
:return: True if successful.
:raises ClientError: If the topic deletion fails.
"""
try:
self.sns_client.delete_topic(TopicArn=topic_arn)
logger.info(f"Deleted topic: {topic_arn}")
return True
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
if error_code == 'NotFound':
logger.warning(f"Topic not found: {topic_arn}")
return True # Already deleted
else:
logger.error(f"Error deleting topic: {error_code} - {e}")
raise
def list_topics(self) -> list:
"""
List all SNS topics in the account using pagination.
:return: List of topic ARNs.
:raises ClientError: If listing topics fails.
"""
try:
topics = []
paginator = self.sns_client.get_paginator('list_topics')
for page in paginator.paginate():
topics.extend([topic['TopicArn'] for topic in page.get('Topics', [])])
logger.info(f"Found {len(topics)} topics")
return topics
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
if error_code == 'AuthorizationError':
logger.error("Authorization error listing topics - check IAM permissions")
else:
logger.error(f"Error listing topics: {error_code} - {e}")
raise
class SqsWrapper:
"""Wrapper class for managing Amazon SQS operations."""
def __init__(self, sqs_client: Any) -> None:
"""
Initialize the SqsWrapper.
:param sqs_client: A Boto3 Amazon SQS client.
"""
self.sqs_client = sqs_client
@classmethod
def from_client(cls) -> 'SqsWrapper':
"""
Create an SqsWrapper instance using a default boto3 client.
:return: An instance of this class.
"""
sqs_client = boto3.client('sqs')
return cls(sqs_client)
def create_queue(self, queue_name: str, is_fifo: bool = False) -> str:
"""
Create an SQS queue.
:param queue_name: The name of the queue to create.
:param is_fifo: Whether to create a FIFO queue.
:return: The URL of the created queue.
:raises ClientError: If the queue creation fails.
"""
try:
# Add .fifo suffix for FIFO queues
if is_fifo and not queue_name.endswith('.fifo'):
queue_name += '.fifo'
attributes = {}
if is_fifo:
attributes['FifoQueue'] = 'true'
response = self.sqs_client.create_queue(
QueueName=queue_name,
Attributes=attributes
)
queue_url = response['QueueUrl']
logger.info(f"Created queue: {queue_name} with URL: {queue_url}")
return queue_url
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error creating queue {queue_name}: {error_code} - {e}")
raise
def get_queue_arn(self, queue_url: str) -> str:
"""
Get the ARN of an SQS queue.
:param queue_url: The URL of the queue.
:return: The ARN of the queue.
:raises ClientError: If getting queue attributes fails.
"""
try:
response = self.sqs_client.get_queue_attributes(
QueueUrl=queue_url,
AttributeNames=['QueueArn']
)
queue_arn = response['Attributes']['QueueArn']
logger.info(f"Queue ARN for {queue_url}: {queue_arn}")
return queue_arn
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error getting queue ARN: {error_code} - {e}")
raise
def set_queue_policy_for_topic(self, queue_arn: str, topic_arn: str, queue_url: str) -> bool:
"""
Set the queue policy to allow SNS to send messages to the queue.
:param queue_arn: The ARN of the SQS queue.
:param topic_arn: The ARN of the SNS topic.
:param queue_url: The URL of the SQS queue.
:return: True if successful.
:raises ClientError: If setting the queue policy fails.
"""
try:
# Create policy that allows SNS to send messages to the queue
policy = {
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "sns.amazonaws.com"
},
"Action": "sqs:SendMessage",
"Resource": queue_arn,
"Condition": {
"ArnEquals": {
"aws:SourceArn": topic_arn
}
}
}
]
}
self.sqs_client.set_queue_attributes(
QueueUrl=queue_url,
Attributes={
'Policy': json.dumps(policy)
}
)
logger.info(f"Set queue policy for {queue_url} to allow messages from {topic_arn}")
return True
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error setting queue policy: {error_code} - {e}")
raise
def receive_messages(self, queue_url: str, max_messages: int = 10) -> List[Dict[str, Any]]:
"""
Receive messages from an SQS queue.
:param queue_url: The URL of the queue to receive messages from.
:param max_messages: Maximum number of messages to receive (1-10).
:return: List of received messages.
:raises ClientError: If receiving messages fails.
"""
try:
# Ensure max_messages is within valid range
max_messages = max(1, min(10, max_messages))
response = self.sqs_client.receive_message(
QueueUrl=queue_url,
MaxNumberOfMessages=max_messages,
WaitTimeSeconds=2, # Short polling
MessageAttributeNames=['All']
)
messages = response.get('Messages', [])
logger.info(f"Received {len(messages)} messages from {queue_url}")
return messages
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error receiving messages: {error_code} - {e}")
raise
def delete_messages(self, queue_url: str, messages: List[Dict[str, Any]]) -> bool:
"""
Delete messages from an SQS queue in batches.
:param queue_url: The URL of the queue.
:param messages: List of messages to delete.
:return: True if successful.
:raises ClientError: If deleting messages fails.
"""
try:
if not messages:
return True
# Build delete entries for batch delete
delete_entries = []
for i, message in enumerate(messages):
delete_entries.append({
'Id': str(i),
'ReceiptHandle': message['ReceiptHandle']
})
# Delete messages in batches of 10 (SQS limit)
batch_size = 10
for i in range(0, len(delete_entries), batch_size):
batch = delete_entries[i:i + batch_size]
response = self.sqs_client.delete_message_batch(
QueueUrl=queue_url,
Entries=batch
)
# Check for failures
if 'Failed' in response and response['Failed']:
for failed in response['Failed']:
logger.warning(f"Failed to delete message: {failed}")
logger.info(f"Deleted {len(messages)} messages from {queue_url}")
return True
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error deleting messages: {error_code} - {e}")
raise
def delete_queue(self, queue_url: str) -> bool:
"""
Delete an SQS queue.
:param queue_url: The URL of the queue to delete.
:return: True if successful.
:raises ClientError: If the queue deletion fails.
"""
try:
self.sqs_client.delete_queue(QueueUrl=queue_url)
logger.info(f"Deleted queue: {queue_url}")
return True
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
if error_code == 'AWS.SimpleQueueService.NonExistentQueue':
logger.warning(f"Queue not found: {queue_url}")
return True # Already deleted
else:
logger.error(f"Error deleting queue: {error_code} - {e}")
raise
def list_queues(self, queue_name_prefix: Optional[str] = None) -> List[str]:
"""
List all SQS queues in the account using pagination.
:param queue_name_prefix: Optional prefix to filter queue names.
:return: List of queue URLs.
:raises ClientError: If listing queues fails.
"""
try:
queue_urls = []
paginator = self.sqs_client.get_paginator('list_queues')
page_params = {}
if queue_name_prefix:
page_params['QueueNamePrefix'] = queue_name_prefix
for page in paginator.paginate(**page_params):
queue_urls.extend(page.get('QueueUrls', []))
logger.info(f"Found {len(queue_urls)} queues")
return queue_urls
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
if error_code == 'AccessDenied':
logger.error("Access denied listing queues - check IAM permissions")
else:
logger.error(f"Error listing queues: {error_code} - {e}")
raise
def send_message(self, queue_url: str, message_body: str, **kwargs) -> str:
"""
Send a message to an SQS queue.
:param queue_url: The URL of the queue.
:param message_body: The message content.
:param kwargs: Additional message parameters (DelaySeconds, MessageAttributes, etc.).
:return: The message ID.
:raises ClientError: If sending the message fails.
"""
try:
send_params = {
'QueueUrl': queue_url,
'MessageBody': message_body,
**kwargs
}
response = self.sqs_client.send_message(**send_params)
message_id = response['MessageId']
logger.info(f"Sent message to {queue_url} with ID: {message_id}")
return message_id
except ClientError as e:
error_code = e.response.get('Error', {}).get('Code', 'Unknown')
logger.error(f"Error sending message: {error_code} - {e}")
raise
```
+ For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/CreateQueue)
+ [CreateTopic](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/CreateTopic)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteQueue)
+ [DeleteTopic](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/DeleteTopic)
+ [GetQueueAttributes](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/GetQueueAttributes)
+ [Publish](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Publish)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/ReceiveMessage)
+ [SetQueueAttributes](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SetQueueAttributes)
+ [Subscribe](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Subscribe)
+ [Unsubscribe](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Unsubscribe)
------
#### [ Swift ]
**SDK for Swift**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/swift/example_code/sqs/scenario#code-examples).
```
import ArgumentParser
import AWSClientRuntime
import AWSSNS
import AWSSQS
import Foundation
struct ExampleCommand: ParsableCommand {
@Option(help: "Name of the Amazon Region to use")
var region = "us-east-1"
static var configuration = CommandConfiguration(
commandName: "queue-scenario",
abstract: """
This example interactively demonstrates how to use Amazon Simple
Notification Service (Amazon SNS) and Amazon Simple Queue Service
(Amazon SQS) together to publish and receive messages using queues.
""",
discussion: """
Supports filtering using a "tone" attribute.
"""
)
/// Prompt for an input string. Only non-empty strings are allowed.
///
/// - Parameter prompt: The prompt to display.
///
/// - Returns: The string input by the user.
func stringRequest(prompt: String) -> String {
var str: String?
while str == nil {
print(prompt, terminator: "")
str = readLine()
if str != nil && str?.count == 0 {
str = nil
}
}
return str!
}
/// Ask a yes/no question.
///
/// - Parameter prompt: A prompt string to print.
///
/// - Returns: `true` if the user answered "Y", otherwise `false`.
func yesNoRequest(prompt: String) -> Bool {
while true {
let answer = stringRequest(prompt: prompt).lowercased()
if answer == "y" || answer == "n" {
return answer == "y"
}
}
}
/// Display a menu of options then request a selection.
///
/// - Parameters:
/// - prompt: A prompt string to display before the menu.
/// - options: An array of strings giving the menu options.
///
/// - Returns: The index number of the selected option or 0 if no item was
/// selected.
func menuRequest(prompt: String, options: [String]) -> Int {
let numOptions = options.count
if numOptions == 0 {
return 0
}
print(prompt)
for (index, value) in options.enumerated() {
print("(\(index)) \(value)")
}
repeat {
print("Enter your selection (0 - \(numOptions-1)): ", terminator: "")
if let answer = readLine() {
guard let answer = Int(answer) else {
print("Please enter the number matching your selection.")
continue
}
if answer >= 0 && answer < numOptions {
return answer
} else {
print("Please enter the number matching your selection.")
}
}
} while true
}
/// Ask the user too press RETURN. Accepts any input but ignores it.
///
/// - Parameter prompt: The text prompt to display.
func returnRequest(prompt: String) {
print(prompt, terminator: "")
_ = readLine()
}
var attrValues = [
"",
"cheerful",
"funny",
"serious",
"sincere"
]
/// Ask the user to choose one of the attribute values to use as a filter.
///
/// - Parameters:
/// - message: A message to display before the menu of values.
/// - attrValues: An array of strings giving the values to choose from.
///
/// - Returns: The string corresponding to the selected option.
func askForFilter(message: String, attrValues: [String]) -> String? {
print(message)
for (index, value) in attrValues.enumerated() {
print(" [\(index)] \(value)")
}
var answer: Int?
repeat {
answer = Int(stringRequest(prompt: "Select an value for the 'tone' attribute or 0 to end: "))
} while answer == nil || answer! < 0 || answer! > attrValues.count + 1
if answer == 0 {
return nil
}
return attrValues[answer!]
}
/// Prompts the user for filter terms and constructs the attribute
/// record that specifies them.
///
/// - Returns: A mapping of "FilterPolicy" to a JSON string representing
/// the user-defined filter.
func buildFilterAttributes() -> [String:String] {
var attr: [String:String] = [:]
var filterString = ""
var first = true
while let ans = askForFilter(message: "Choose a value to apply to the 'tone' attribute.",
attrValues: attrValues) {
if !first {
filterString += ","
}
first = false
filterString += "\"\(ans)\""
}
let filterJSON = "{ \"tone\": [\(filterString)]}"
attr["FilterPolicy"] = filterJSON
return attr
}
/// Create a queue, returning its URL string.
///
/// - Parameters:
/// - prompt: A prompt to ask for the queue name.
/// - isFIFO: Whether or not to create a FIFO queue.
///
/// - Returns: The URL of the queue.
func createQueue(prompt: String, sqsClient: SQSClient, isFIFO: Bool) async throws -> String? {
repeat {
var queueName = stringRequest(prompt: prompt)
var attributes: [String: String] = [:]
if isFIFO {
queueName += ".fifo"
attributes["FifoQueue"] = "true"
}
do {
let output = try await sqsClient.createQueue(
input: CreateQueueInput(
attributes: attributes,
queueName: queueName
)
)
guard let url = output.queueUrl else {
return nil
}
return url
} catch _ as QueueDeletedRecently {
print("You need to use a different queue name. A queue by that name was recently deleted.")
continue
}
} while true
}
/// Return the ARN of a queue given its URL.
///
/// - Parameter queueUrl: The URL of the queue for which to return the
/// ARN.
///
/// - Returns: The ARN of the specified queue.
func getQueueARN(sqsClient: SQSClient, queueUrl: String) async throws -> String? {
let output = try await sqsClient.getQueueAttributes(
input: GetQueueAttributesInput(
attributeNames: [.queuearn],
queueUrl: queueUrl
)
)
guard let attributes = output.attributes else {
return nil
}
return attributes["QueueArn"]
}
/// Applies the needed policy to the specified queue.
///
/// - Parameters:
/// - sqsClient: The Amazon SQS client to use.
/// - queueUrl: The queue to apply the policy to.
/// - queueArn: The ARN of the queue to apply the policy to.
/// - topicArn: The topic that should have access via the policy.
///
/// - Throws: Errors from the SQS `SetQueueAttributes` action.
func setQueuePolicy(sqsClient: SQSClient, queueUrl: String,
queueArn: String, topicArn: String) async throws {
_ = try await sqsClient.setQueueAttributes(
input: SetQueueAttributesInput(
attributes: [
"Policy":
"""
{
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "sns.amazonaws.com"
},
"Action": "sqs:SendMessage",
"Resource": "\(queueArn)",
"Condition": {
"ArnEquals": {
"aws:SourceArn": "\(topicArn)"
}
}
}
]
}
"""
],
queueUrl: queueUrl
)
)
}
/// Receive the available messages on a queue, outputting them to the
/// screen. Returns a dictionary you pass to DeleteMessageBatch to delete
/// all the received messages.
///
/// - Parameters:
/// - sqsClient: The Amazon SQS client to use.
/// - queueUrl: The SQS queue on which to receive messages.
///
/// - Throws: Errors from `SQSClient.receiveMessage()`
///
/// - Returns: An array of SQSClientTypes.DeleteMessageBatchRequestEntry
/// items, each describing one received message in the format needed to
/// delete it.
func receiveAndListMessages(sqsClient: SQSClient, queueUrl: String) async throws
-> [SQSClientTypes.DeleteMessageBatchRequestEntry] {
let output = try await sqsClient.receiveMessage(
input: ReceiveMessageInput(
maxNumberOfMessages: 10,
queueUrl: queueUrl
)
)
guard let messages = output.messages else {
print("No messages received.")
return []
}
var deleteList: [SQSClientTypes.DeleteMessageBatchRequestEntry] = []
// Print out all the messages that were received, including their
// attributes, if any.
for message in messages {
print("Message ID: \(message.messageId ?? "")")
print("Receipt handle: \(message.receiptHandle ?? "")")
print("Message JSON: \(message.body ?? "")")
if message.receiptHandle != nil {
deleteList.append(
SQSClientTypes.DeleteMessageBatchRequestEntry(
id: message.messageId,
receiptHandle: message.receiptHandle
)
)
}
}
return deleteList
}
/// Delete all the messages in the specified list.
///
/// - Parameters:
/// - sqsClient: The Amazon SQS client to use.
/// - queueUrl: The SQS queue to delete messages from.
/// - deleteList: A list of `DeleteMessageBatchRequestEntry` objects
/// describing the messages to delete.
///
/// - Throws: Errors from `SQSClient.deleteMessageBatch()`.
func deleteMessageList(sqsClient: SQSClient, queueUrl: String,
deleteList: [SQSClientTypes.DeleteMessageBatchRequestEntry]) async throws {
let output = try await sqsClient.deleteMessageBatch(
input: DeleteMessageBatchInput(entries: deleteList, queueUrl: queueUrl)
)
if let failed = output.failed {
print("\(failed.count) errors occurred deleting messages from the queue.")
for message in failed {
print("---> Failed to delete message \(message.id ?? "") with error: \(message.code ?? "") (\(message.message ?? "..."))")
}
}
}
/// Called by ``main()`` to run the bulk of the example.
func runAsync() async throws {
let rowOfStars = String(repeating: "*", count: 75)
print("""
\(rowOfStars)
Welcome to the cross-service messaging with topics and queues example.
In this workflow, you'll create an SNS topic, then create two SQS
queues which will be subscribed to that topic.
You can specify several options for configuring the topic, as well as
the queue subscriptions. You can then post messages to the topic and
receive the results on the queues.
\(rowOfStars)\n
"""
)
// 0. Create SNS and SQS clients.
let snsConfig = try await SNSClient.SNSClientConfiguration(region: region)
let snsClient = SNSClient(config: snsConfig)
let sqsConfig = try await SQSClient.SQSClientConfiguration(region: region)
let sqsClient = SQSClient(config: sqsConfig)
// 1. Ask the user whether to create a FIFO topic. If so, ask whether
// to use content-based deduplication instead of requiring a
// deduplication ID.
let isFIFO = yesNoRequest(prompt: "Do you want to create a FIFO topic (Y/N)? ")
var isContentBasedDeduplication = false
if isFIFO {
print("""
\(rowOfStars)
Because you've chosen to create a FIFO topic, deduplication is
supported.
Deduplication IDs are either set in the message or are automatically
generated from the content using a hash function.
If a message is successfully published to an SNS FIFO topic, any
message published and found to have the same deduplication ID
(within a five-minute deduplication interval), is accepted but
not delivered.
For more information about deduplication, see:
https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html.
"""
)
isContentBasedDeduplication = yesNoRequest(
prompt: "Use content-based deduplication instead of entering a deduplication ID (Y/N)? ")
print(rowOfStars)
}
var topicName = stringRequest(prompt: "Enter the name of the topic to create: ")
// 2. Create the topic. Append ".fifo" to the name if FIFO was
// requested, and set the "FifoTopic" attribute to "true" if so as
// well. Set the "ContentBasedDeduplication" attribute to "true" if
// content-based deduplication was requested.
if isFIFO {
topicName += ".fifo"
}
print("Topic name: \(topicName)")
var attributes = [
"FifoTopic": (isFIFO ? "true" : "false")
]
// If it's a FIFO topic with content-based deduplication, set the
// "ContentBasedDeduplication" attribute.
if isContentBasedDeduplication {
attributes["ContentBasedDeduplication"] = "true"
}
// Create the topic and retrieve the ARN.
let output = try await snsClient.createTopic(
input: CreateTopicInput(
attributes: attributes,
name: topicName
)
)
guard let topicArn = output.topicArn else {
print("No topic ARN returned!")
return
}
print("""
Topic '\(topicName) has been created with the
topic ARN \(topicArn)."
"""
)
print(rowOfStars)
// 3. Create an SQS queue. Append ".fifo" to the name if one of the
// FIFO topic configurations was chosen, and set "FifoQueue" to
// "true" if the topic is FIFO.
print("""
Next, you will create two SQS queues that will be subscribed
to the topic you just created.\n
"""
)
let q1Url = try await createQueue(prompt: "Enter the name of the first queue: ",
sqsClient: sqsClient, isFIFO: isFIFO)
guard let q1Url else {
print("Unable to create queue 1!")
return
}
// 4. Get the SQS queue's ARN attribute using `GetQueueAttributes`.
let q1Arn = try await getQueueARN(sqsClient: sqsClient, queueUrl: q1Url)
guard let q1Arn else {
print("Unable to get ARN of queue 1!")
return
}
print("Got queue 1 ARN: \(q1Arn)")
// 5. Attach an AWS IAM policy to the queue using
// `SetQueueAttributes`.
try await setQueuePolicy(sqsClient: sqsClient, queueUrl: q1Url,
queueArn: q1Arn, topicArn: topicArn)
// 6. Subscribe the SQS queue to the SNS topic. Set the topic ARN in
// the request. Set the protocol to "sqs". Set the queue ARN to the
// ARN just received in step 5. For FIFO topics, give the option to
// apply a filter. A filter allows only matching messages to enter
// the queue.
var q1Attributes: [String:String]? = nil
if isFIFO {
print(
"""
If you add a filter to this subscription, then only the filtered messages will
be received in the queue. For information about message filtering, see
https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html
For this example, you can filter messages by a 'tone' attribute.
"""
)
let subPrompt = """
Would you like to filter messages for the first queue's subscription to the
topic \(topicName) (Y/N)?
"""
if (yesNoRequest(prompt: subPrompt)) {
q1Attributes = buildFilterAttributes()
}
}
let sub1Output = try await snsClient.subscribe(
input: SubscribeInput(
attributes: q1Attributes,
endpoint: q1Arn,
protocol: "sqs",
topicArn: topicArn
)
)
guard let q1SubscriptionArn = sub1Output.subscriptionArn else {
print("Invalid subscription ARN returned for queue 1!")
return
}
// 7. Repeat steps 3-6 for the second queue.
let q2Url = try await createQueue(prompt: "Enter the name of the second queue: ",
sqsClient: sqsClient, isFIFO: isFIFO)
guard let q2Url else {
print("Unable to create queue 2!")
return
}
let q2Arn = try await getQueueARN(sqsClient: sqsClient, queueUrl: q2Url)
guard let q2Arn else {
print("Unable to get ARN of queue 2!")
return
}
print("Got queue 2 ARN: \(q2Arn)")
try await setQueuePolicy(sqsClient: sqsClient, queueUrl: q2Url,
queueArn: q2Arn, topicArn: topicArn)
var q2Attributes: [String:String]? = nil
if isFIFO {
let subPrompt = """
Would you like to filter messages for the second queue's subscription to the
topic \(topicName) (Y/N)?
"""
if (yesNoRequest(prompt: subPrompt)) {
q2Attributes = buildFilterAttributes()
}
}
let sub2Output = try await snsClient.subscribe(
input: SubscribeInput(
attributes: q2Attributes,
endpoint: q2Arn,
protocol: "sqs",
topicArn: topicArn
)
)
guard let q2SubscriptionArn = sub2Output.subscriptionArn else {
print("Invalid subscription ARN returned for queue 1!")
return
}
// 8. Let the user publish messages to the topic, asking for a message
// body for each message. Handle the types of topic correctly (SEE
// MVP INFORMATION AND FIX THESE COMMENTS!!!
print("\n\(rowOfStars)\n")
var first = true
repeat {
var publishInput = PublishInput(
topicArn: topicArn
)
publishInput.message = stringRequest(prompt: "Enter message text to publish: ")
// If using a FIFO topic, a message group ID must be set on the
// message.
if isFIFO {
if first {
print("""
Because you're using a FIFO topic, you must set a message
group ID. All messages within the same group will be
received in the same order in which they were published.\n
"""
)
}
publishInput.messageGroupId = stringRequest(prompt: "Enter a message group ID for this message: ")
if !isContentBasedDeduplication {
if first {
print("""
Because you're not using content-based deduplication, you
must enter a deduplication ID. If other messages with the
same deduplication ID are published within the same
deduplication interval, they will not be delivered.
"""
)
}
publishInput.messageDeduplicationId = stringRequest(prompt: "Enter a deduplication ID for this message: ")
}
}
// Allow the user to add a value for the "tone" attribute if they
// wish to do so.
var messageAttributes: [String:SNSClientTypes.MessageAttributeValue] = [:]
let attrValSelection = menuRequest(prompt: "Choose a tone to apply to this message.", options: attrValues)
if attrValSelection != 0 {
let val = SNSClientTypes.MessageAttributeValue(dataType: "String", stringValue: attrValues[attrValSelection])
messageAttributes["tone"] = val
}
publishInput.messageAttributes = messageAttributes
// Publish the message and display its ID.
let publishOutput = try await snsClient.publish(input: publishInput)
guard let messageID = publishOutput.messageId else {
print("Unable to get the published message's ID!")
return
}
print("Message published with ID \(messageID).")
first = false
// 9. Repeat step 8 until the user says they don't want to post
// another.
} while (yesNoRequest(prompt: "Post another message (Y/N)? "))
// 10. Display a list of the messages in each queue by using
// `ReceiveMessage`. Show at least the body and the attributes.
print(rowOfStars)
print("Contents of queue 1:")
let q1DeleteList = try await receiveAndListMessages(sqsClient: sqsClient, queueUrl: q1Url)
print("\n\nContents of queue 2:")
let q2DeleteList = try await receiveAndListMessages(sqsClient: sqsClient, queueUrl: q2Url)
print(rowOfStars)
returnRequest(prompt: "\nPress return to clean up: ")
// 11. Delete the received messages using `DeleteMessageBatch`.
print("Deleting the messages from queue 1...")
try await deleteMessageList(sqsClient: sqsClient, queueUrl: q1Url, deleteList: q1DeleteList)
print("\nDeleting the messages from queue 2...")
try await deleteMessageList(sqsClient: sqsClient, queueUrl: q2Url, deleteList: q2DeleteList)
// 12. Unsubscribe and delete both queues.
print("\nUnsubscribing from queue 1...")
_ = try await snsClient.unsubscribe(
input: UnsubscribeInput(subscriptionArn: q1SubscriptionArn)
)
print("Unsubscribing from queue 2...")
_ = try await snsClient.unsubscribe(
input: UnsubscribeInput(subscriptionArn: q2SubscriptionArn)
)
print("Deleting queue 1...")
_ = try await sqsClient.deleteQueue(
input: DeleteQueueInput(queueUrl: q1Url)
)
print("Deleting queue 2...")
_ = try await sqsClient.deleteQueue(
input: DeleteQueueInput(queueUrl: q2Url)
)
// 13. Delete the topic.
print("Deleting the SNS topic...")
_ = try await snsClient.deleteTopic(
input: DeleteTopicInput(topicArn: topicArn)
)
}
}
/// The program's asynchronous entry point.
@main
struct Main {
static func main() async {
let args = Array(CommandLine.arguments.dropFirst())
do {
let command = try ExampleCommand.parse(args)
try await command.runAsync()
} catch {
ExampleCommand.exit(withError: error)
}
}
}
```
+ For API details, see the following topics in *AWS SDK for Swift API reference*.
+ [CreateQueue](https://sdk.amazonaws.com/swift/api/awssqs/latest/documentation/awssqs/sqsclient/createqueue(input:))
+ [CreateTopic](https://sdk.amazonaws.com/swift/api/awssns/latest/documentation/awssns/snsclient/createtopic(input:))
+ [DeleteMessageBatch](https://sdk.amazonaws.com/swift/api/awssqs/latest/documentation/awssqs/sqsclient/deletemessagebatch(input:))
+ [DeleteQueue](https://sdk.amazonaws.com/swift/api/awssqs/latest/documentation/awssqs/sqsclient/deletequeue(input:))
+ [DeleteTopic](https://sdk.amazonaws.com/swift/api/awssns/latest/documentation/awssns/snsclient/deletetopic(input:))
+ [GetQueueAttributes](https://sdk.amazonaws.com/swift/api/awssqs/latest/documentation/awssqs/sqsclient/getqueueattributes(input:))
+ [Publish](https://sdk.amazonaws.com/swift/api/awssns/latest/documentation/awssns/snsclient/publish(input:))
+ [ReceiveMessage](https://sdk.amazonaws.com/swift/api/awssqs/latest/documentation/awssqs/sqsclient/receivemessage(input:))
+ [SetQueueAttributes](https://sdk.amazonaws.com/swift/api/awssqs/latest/documentation/awssqs/sqsclient/setqueueattributes(input:))
+ [Subscribe](https://sdk.amazonaws.com/swift/api/awssns/latest/documentation/awssns/snsclient/subscribe(input:))
+ [Unsubscribe](https://sdk.amazonaws.com/swift/api/awssns/latest/documentation/awssns/snsclient/unsubscribe(input:))
------
# Send and receive batches of messages with Amazon SQS using an AWS SDK
The following code examples show how to:
+ Create an Amazon SQS queue.
+ Send batches of messages to the queue.
+ Receive batches of messages from the queue.
+ Delete batches of messages from the queue.
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sqs#code-examples).
As shown in the following examples, you can handle batch message operations with Amazon SQS using two different approaches with the AWS SDK for Java 2.x:
**SendRecvBatch.java** uses explicit batch operations. You manually create message batches and call `sendMessageBatch()` and `deleteMessageBatch()` directly. You also handle batch responses, including any failed messages. This approach gives you full control over batch sizing and error handling. However, it requires more code to manage the batching logic.
**SimpleProducerConsumer.java** uses the high-level `SqsAsyncBatchManager` library for automatic request batching. You make individual `sendMessage()` and `deleteMessage()` calls with the same method signatures as the standard client. The SDK automatically buffers these calls and sends them as batch operations. This approach requires minimal code changes while providing batching performance benefits.
Use explicit batching when you need fine-grained control over batch composition and error handling. Use automatic batching when you want to optimize performance with minimal code changes.
SendRecvBatch.java - Uses explicit batch operations with messages.
```
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
import software.amazon.awssdk.services.sqs.model.CreateQueueRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResponse;
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResultEntry;
import software.amazon.awssdk.services.sqs.model.DeleteQueueRequest;
import software.amazon.awssdk.services.sqs.model.Message;
import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest;
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;
import software.amazon.awssdk.services.sqs.model.SendMessageBatchResponse;
import software.amazon.awssdk.services.sqs.model.SendMessageBatchResultEntry;
import software.amazon.awssdk.services.sqs.model.SqsException;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
/**
* This code demonstrates basic message operations in Amazon Simple Queue Service (Amazon SQS).
*/
public class SendRecvBatch {
private static final Logger LOGGER = LoggerFactory.getLogger(SendRecvBatch.class);
private static final SqsClient sqsClient = SqsClient.create();
public static void main(String[] args) {
usageDemo();
}
/**
* Send a batch of messages in a single request to an SQS queue.
* This request may return overall success even when some messages were not sent.
* The caller must inspect the Successful and Failed lists in the response and
* resend any failed messages.
*
* @param queueUrl The URL of the queue to receive the messages.
* @param messages The messages to send to the queue. Each message contains a body and attributes.
* @return The response from SQS that contains the list of successful and failed messages.
*/
public static SendMessageBatchResponse sendMessages(
String queueUrl, List messages) {
try {
List entries = new ArrayList<>();
for (int i = 0; i < messages.size(); i++) {
MessageEntry message = messages.get(i);
entries.add(SendMessageBatchRequestEntry.builder()
.id(String.valueOf(i))
.messageBody(message.getBody())
.messageAttributes(message.getAttributes())
.build());
}
SendMessageBatchRequest sendBatchRequest = SendMessageBatchRequest.builder()
.queueUrl(queueUrl)
.entries(entries)
.build();
SendMessageBatchResponse response = sqsClient.sendMessageBatch(sendBatchRequest);
if (!response.successful().isEmpty()) {
for (SendMessageBatchResultEntry resultEntry : response.successful()) {
LOGGER.info("Message sent: {}: {}", resultEntry.messageId(),
messages.get(Integer.parseInt(resultEntry.id())).getBody());
}
}
if (!response.failed().isEmpty()) {
for (BatchResultErrorEntry errorEntry : response.failed()) {
LOGGER.warn("Failed to send: {}: {}", errorEntry.id(),
messages.get(Integer.parseInt(errorEntry.id())).getBody());
}
}
return response;
} catch (SqsException e) {
LOGGER.error("Send messages failed to queue: {}", queueUrl, e);
throw e;
}
}
/**
* Receive a batch of messages in a single request from an SQS queue.
*
* @param queueUrl The URL of the queue from which to receive messages.
* @param maxNumber The maximum number of messages to receive (capped at 10 by SQS).
* The actual number of messages received might be less.
* @param waitTime The maximum time to wait (in seconds) before returning. When
* this number is greater than zero, long polling is used. This
* can result in reduced costs and fewer false empty responses.
* @return The list of Message objects received. These each contain the body
* of the message and metadata and custom attributes.
*/
public static List receiveMessages(String queueUrl, int maxNumber, int waitTime) {
try {
ReceiveMessageRequest receiveRequest = ReceiveMessageRequest.builder()
.queueUrl(queueUrl)
.maxNumberOfMessages(maxNumber)
.waitTimeSeconds(waitTime)
.messageAttributeNames("All")
.build();
List messages = sqsClient.receiveMessage(receiveRequest).messages();
for (Message message : messages) {
LOGGER.info("Received message: {}: {}", message.messageId(), message.body());
}
return messages;
} catch (SqsException e) {
LOGGER.error("Couldn't receive messages from queue: {}", queueUrl, e);
throw e;
}
}
/**
* Delete a batch of messages from a queue in a single request.
*
* @param queueUrl The URL of the queue from which to delete the messages.
* @param messages The list of messages to delete.
* @return The response from SQS that contains the list of successful and failed
* message deletions.
*/
public static DeleteMessageBatchResponse deleteMessages(String queueUrl, List messages) {
try {
List entries = new ArrayList<>();
for (int i = 0; i < messages.size(); i++) {
entries.add(DeleteMessageBatchRequestEntry.builder()
.id(String.valueOf(i))
.receiptHandle(messages.get(i).receiptHandle())
.build());
}
DeleteMessageBatchRequest deleteRequest = DeleteMessageBatchRequest.builder()
.queueUrl(queueUrl)
.entries(entries)
.build();
DeleteMessageBatchResponse response = sqsClient.deleteMessageBatch(deleteRequest);
if (!response.successful().isEmpty()) {
for (DeleteMessageBatchResultEntry resultEntry : response.successful()) {
LOGGER.info("Deleted {}", messages.get(Integer.parseInt(resultEntry.id())).receiptHandle());
}
}
if (!response.failed().isEmpty()) {
for (BatchResultErrorEntry errorEntry : response.failed()) {
LOGGER.warn("Could not delete {}", messages.get(Integer.parseInt(errorEntry.id())).receiptHandle());
}
}
return response;
} catch (SqsException e) {
LOGGER.error("Couldn't delete messages from queue {}", queueUrl, e);
throw e;
}
}
/**
* Helper class to represent a message with body and attributes.
*/
public static class MessageEntry {
private final String body;
private final Map attributes;
public MessageEntry(String body, Map attributes) {
this.body = body;
this.attributes = attributes != null ? attributes : new HashMap<>();
}
public String getBody() {
return body;
}
public Map getAttributes() {
return attributes;
}
}
/**
* Shows how to:
* * Read the lines from a file and send the lines in
* batches of 10 as messages to a queue.
* * Receive the messages in batches until the queue is empty.
* * Reassemble the lines of the file and verify they match the original file.
*/
public static void usageDemo() {
LOGGER.info("-".repeat(88));
LOGGER.info("Welcome to the Amazon Simple Queue Service (Amazon SQS) demo!");
LOGGER.info("-".repeat(88));
String queueUrl = null;
try {
// Create a queue for the demo.
String queueName = "sqs-usage-demo-message-wrapper-" + System.currentTimeMillis();
CreateQueueRequest createRequest = CreateQueueRequest.builder()
.queueName(queueName)
.build();
queueUrl = sqsClient.createQueue(createRequest).queueUrl();
LOGGER.info("Created queue: {}", queueUrl);
try (InputStream inputStream = SendRecvBatch.class.getResourceAsStream("/log4j2.xml");
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
List lines = reader.lines().toList();
// Send file lines in batches.
int batchSize = 10;
LOGGER.info("Sending file lines in batches of {} as messages.", batchSize);
for (int i = 0; i < lines.size(); i += batchSize) {
List messageBatch = new ArrayList<>();
for (int j = i; j < Math.min(i + batchSize, lines.size()); j++) {
String line = lines.get(j);
if (line == null || line.trim().isEmpty()) {
continue; // Skip empty lines.
}
Map attributes = new HashMap<>();
attributes.put("line", MessageAttributeValue.builder()
.dataType("String")
.stringValue(String.valueOf(j))
.build());
messageBatch.add(new MessageEntry(lines.get(j), attributes));
}
sendMessages(queueUrl, messageBatch);
System.out.print(".");
System.out.flush();
}
LOGGER.info("\nDone. Sent {} messages.", lines.size());
// Receive and process messages.
LOGGER.info("Receiving, handling, and deleting messages in batches of {}.", batchSize);
String[] receivedLines = new String[lines.size()];
boolean moreMessages = true;
while (moreMessages) {
List receivedMessages = receiveMessages(queueUrl, batchSize, 5);
for (Message message : receivedMessages) {
int lineNumber = Integer.parseInt(message.messageAttributes().get("line").stringValue());
receivedLines[lineNumber] = message.body();
}
if (!receivedMessages.isEmpty()) {
deleteMessages(queueUrl, receivedMessages);
} else {
moreMessages = false;
}
}
LOGGER.info("\nDone.");
// Verify that all lines were received correctly.
boolean allLinesMatch = true;
for (int i = 0; i < lines.size(); i++) {
String originalLine = lines.get(i);
String receivedLine = receivedLines[i] == null ? "" : receivedLines[i];
if (!originalLine.equals(receivedLine)) {
allLinesMatch = false;
break;
}
}
if (allLinesMatch) {
LOGGER.info("Successfully reassembled all file lines!");
} else {
LOGGER.info("Uh oh, some lines were missed!");
}
}
} catch (SqsException e) {
LOGGER.error("SQS operation failed", e);
} catch (RuntimeException | IOException e) {
LOGGER.error("Unexpected runtime error during demo", e);
} finally {
// Clean up by deleting the queue if it was created.
if (queueUrl != null) {
try {
DeleteQueueRequest deleteQueueRequest = DeleteQueueRequest.builder()
.queueUrl(queueUrl)
.build();
sqsClient.deleteQueue(deleteQueueRequest);
LOGGER.info("Deleted queue: {}", queueUrl);
} catch (SqsException e) {
LOGGER.error("Failed to delete queue: {}", queueUrl, e);
}
}
}
LOGGER.info("Thanks for watching!");
LOGGER.info("-".repeat(88));
}
}
```
SimpleProducerConsumer.java - Uses automatic batching of messages.
```
package com.example.sqs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
import software.amazon.awssdk.services.sqs.batchmanager.SqsAsyncBatchManager;
import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageResponse;
import software.amazon.awssdk.services.sqs.model.GetQueueUrlRequest;
import software.amazon.awssdk.services.sqs.model.Message;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
import software.amazon.awssdk.services.sqs.model.SendMessageResponse;
import software.amazon.awssdk.core.exception.SdkException;
import java.math.BigInteger;
import java.util.List;
import java.util.Random;
import java.util.Scanner;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Demonstrates the AWS SDK for Java 2.x Automatic Request Batching API for Amazon SQS.
*
* This example showcases the high-level SqsAsyncBatchManager library that provides
* efficient batching and buffering for SQS operations. The batch manager offers
* methods that directly mirror SqsAsyncClient methods—sendMessage, changeMessageVisibility,
* deleteMessage, and receiveMessage—making it a drop-in replacement with minimal code changes.
*
* Key features of the SqsAsyncBatchManager:
* - Automatic batching: The SDK automatically buffers individual requests and sends them
* as batches when maxBatchSize (default: 10) or sendRequestFrequency (default: 200ms)
* thresholds are reached
* - Familiar API: Method signatures match SqsAsyncClient exactly, requiring no learning curve
* - Background optimization: The batch manager maintains internal buffers and handles
* batching logic transparently
* - Asynchronous operations: All methods return CompletableFuture for non-blocking execution
*
* Performance benefits demonstrated:
* - Reduced API calls: Multiple individual requests are consolidated into single batch operations
* - Lower costs: Fewer API calls result in reduced SQS charges
* - Higher throughput: Batch operations process more messages per second
* - Efficient resource utilization: Fewer network round trips and better connection reuse
*
* This example compares:
* 1. Single-message operations using SqsAsyncClient directly
* 2. Batch operations using SqsAsyncBatchManager with identical method calls
*
* Usage patterns:
* - Set batch size to 1 to use SqsAsyncClient for baseline performance measurement
* - Set batch size > 1 to use SqsAsyncBatchManager for optimized batch processing
* - Monitor real-time throughput metrics to observe performance improvements
*
* Prerequisites:
* - AWS SDK for Java 2.x version 2.28.0 or later
* - An existing SQS queue
* - Valid AWS credentials configured
*
* The program displays real-time metrics showing the dramatic performance difference
* between individual operations and automatic batching.
*/
public class SimpleProducerConsumer {
// The maximum runtime of the program.
private final static int MAX_RUNTIME_MINUTES = 60;
private final static Logger log = LoggerFactory.getLogger(SimpleProducerConsumer.class);
/**
* Runs the SQS batching demonstration with user-configured parameters.
*
* Prompts for queue name, thread counts, batch size, message size, and runtime.
* Creates producer and consumer threads to demonstrate batching performance.
*
* @param args command line arguments (not used)
* @throws InterruptedException if thread operations are interrupted
*/
public static void main(String[] args) throws InterruptedException {
final Scanner input = new Scanner(System.in);
System.out.print("Enter the queue name: ");
final String queueName = input.nextLine();
System.out.print("Enter the number of producers: ");
final int producerCount = input.nextInt();
System.out.print("Enter the number of consumers: ");
final int consumerCount = input.nextInt();
System.out.print("Enter the number of messages per batch: ");
final int batchSize = input.nextInt();
System.out.print("Enter the message size in bytes: ");
final int messageSizeByte = input.nextInt();
System.out.print("Enter the run time in minutes: ");
final int runTimeMinutes = input.nextInt();
// Create SQS async client and batch manager for all operations.
// The SqsAsyncBatchManager is created from the SqsAsyncClient using the
// batchManager() factory method, which provides default batching configuration.
// This high-level library automatically handles request buffering and batching
// while maintaining the same method signatures as SqsAsyncClient.
final SqsAsyncClient sqsAsyncClient = SqsAsyncClient.create();
final SqsAsyncBatchManager batchManager = sqsAsyncClient.batchManager();
final String queueUrl = sqsAsyncClient.getQueueUrl(GetQueueUrlRequest.builder()
.queueName(queueName)
.build()).join().queueUrl();
// The flag used to stop producer, consumer, and monitor threads.
final AtomicBoolean stop = new AtomicBoolean(false);
// Start the producers.
final AtomicInteger producedCount = new AtomicInteger();
final Thread[] producers = new Thread[producerCount];
for (int i = 0; i < producerCount; i++) {
if (batchSize == 1) {
producers[i] = new Producer(sqsAsyncClient, queueUrl, messageSizeByte,
producedCount, stop);
} else {
producers[i] = new BatchProducer(batchManager, queueUrl, batchSize,
messageSizeByte, producedCount, stop);
}
producers[i].start();
}
// Start the consumers.
final AtomicInteger consumedCount = new AtomicInteger();
final Thread[] consumers = new Thread[consumerCount];
for (int i = 0; i < consumerCount; i++) {
if (batchSize == 1) {
consumers[i] = new Consumer(sqsAsyncClient, queueUrl, consumedCount, stop);
} else {
consumers[i] = new BatchConsumer(batchManager, queueUrl, batchSize,
consumedCount, stop);
}
consumers[i].start();
}
// Start the monitor thread.
final Thread monitor = new Monitor(producedCount, consumedCount, stop);
monitor.start();
// Wait for the specified amount of time then stop.
Thread.sleep(TimeUnit.MINUTES.toMillis(Math.min(runTimeMinutes,
MAX_RUNTIME_MINUTES)));
stop.set(true);
// Join all threads.
for (int i = 0; i < producerCount; i++) {
producers[i].join();
}
for (int i = 0; i < consumerCount; i++) {
consumers[i].join();
}
monitor.interrupt();
monitor.join();
// Close resources
batchManager.close();
sqsAsyncClient.close();
}
/**
* Creates a random string of approximately the specified size in bytes.
*
* @param sizeByte the target size in bytes for the generated string
* @return a random string encoded in base-32
*/
private static String makeRandomString(int sizeByte) {
final byte[] bs = new byte[(int) Math.ceil(sizeByte * 5 / 8)];
new Random().nextBytes(bs);
bs[0] = (byte) ((bs[0] | 64) & 127);
return new BigInteger(bs).toString(32);
}
/**
* Sends messages individually using SqsAsyncClient for baseline performance measurement.
*
* This producer demonstrates traditional single-message operations without batching.
* Each sendMessage() call results in a separate API request to SQS, providing
* a performance baseline for comparison with the batch operations.
*
* The sendMessage() method signature is identical to SqsAsyncBatchManager.sendMessage(),
* showing how the high-level batching library maintains API compatibility while
* adding automatic optimization behind the scenes.
*/
private static class Producer extends Thread {
final SqsAsyncClient sqsAsyncClient;
final String queueUrl;
final AtomicInteger producedCount;
final AtomicBoolean stop;
final String theMessage;
/**
* Creates a producer thread for single-message operations.
*
* @param sqsAsyncClient the SQS client for sending messages
* @param queueUrl the URL of the target queue
* @param messageSizeByte the size of messages to generate
* @param producedCount shared counter for tracking sent messages
* @param stop shared flag to signal thread termination
*/
Producer(SqsAsyncClient sqsAsyncClient, String queueUrl, int messageSizeByte,
AtomicInteger producedCount, AtomicBoolean stop) {
this.sqsAsyncClient = sqsAsyncClient;
this.queueUrl = queueUrl;
this.producedCount = producedCount;
this.stop = stop;
this.theMessage = makeRandomString(messageSizeByte);
}
/**
* Continuously sends messages until the stop flag is set.
*
* Uses SqsAsyncClient.sendMessage() directly, resulting in one API call per message.
* This approach provides baseline performance metrics for comparison with batching.
* Each call blocks until the individual message is sent, demonstrating traditional
* one-request-per-operation behavior.
*/
public void run() {
try {
while (!stop.get()) {
sqsAsyncClient.sendMessage(SendMessageRequest.builder()
.queueUrl(queueUrl)
.messageBody(theMessage)
.build()).join();
producedCount.incrementAndGet();
}
} catch (SdkException | java.util.concurrent.CompletionException e) {
// Handle both SdkException and CompletionException from async operations.
// If this unlikely condition occurs, stop.
log.error("Producer: " + e.getMessage());
System.exit(1);
}
}
}
/**
* Sends messages using SqsAsyncBatchManager for automatic request batching and optimization.
*
* This producer demonstrates the AWS SDK for Java 2.x high-level batching library.
* The SqsAsyncBatchManager automatically buffers individual sendMessage() calls and
* sends them as batches when thresholds are reached:
* - maxBatchSize: Maximum 10 messages per batch (default)
* - sendRequestFrequency: 200ms timeout before sending partial batches (default)
*
* Key advantages of the batching approach:
* - Identical API: batchManager.sendMessage() has the same signature as sqsAsyncClient.sendMessage()
* - Automatic optimization: No code changes needed to benefit from batching
* - Transparent buffering: The SDK handles batching logic internally
* - Reduced API calls: Multiple messages sent in single batch requests
* - Lower costs: Fewer API calls result in reduced SQS charges
* - Higher throughput: Batch operations process significantly more messages per second
*/
private static class BatchProducer extends Thread {
final SqsAsyncBatchManager batchManager;
final String queueUrl;
final int batchSize;
final AtomicInteger producedCount;
final AtomicBoolean stop;
final String theMessage;
/**
* Creates a producer thread for batch operations.
*
* @param batchManager the batch manager for efficient message sending
* @param queueUrl the URL of the target queue
* @param batchSize the number of messages to send per batch
* @param messageSizeByte the size of messages to generate
* @param producedCount shared counter for tracking sent messages
* @param stop shared flag to signal thread termination
*/
BatchProducer(SqsAsyncBatchManager batchManager, String queueUrl, int batchSize,
int messageSizeByte, AtomicInteger producedCount,
AtomicBoolean stop) {
this.batchManager = batchManager;
this.queueUrl = queueUrl;
this.batchSize = batchSize;
this.producedCount = producedCount;
this.stop = stop;
this.theMessage = makeRandomString(messageSizeByte);
}
/**
* Continuously sends batches of messages using the high-level batching library.
*
* Notice how batchManager.sendMessage() uses the exact same method signature
* and request builder pattern as SqsAsyncClient.sendMessage(). This demonstrates
* the drop-in replacement capability of the SqsAsyncBatchManager.
*
* The SDK automatically:
* - Buffers individual sendMessage() calls internally
* - Groups them into batch requests when thresholds are met
* - Sends SendMessageBatchRequest operations to SQS
* - Returns individual CompletableFuture responses for each message
*
* This transparent batching provides significant performance improvements
* without requiring changes to application logic or error handling patterns.
*/
public void run() {
try {
while (!stop.get()) {
// Send multiple messages using the high-level batch manager.
// Each batchManager.sendMessage() call uses identical syntax to
// sqsAsyncClient.sendMessage(), demonstrating API compatibility.
// The SDK automatically buffers these calls and sends them as
// batch operations when maxBatchSize (10) or sendRequestFrequency (200ms)
// thresholds are reached, significantly improving throughput.
for (int i = 0; i < batchSize; i++) {
CompletableFuture future = batchManager.sendMessage(
SendMessageRequest.builder()
.queueUrl(queueUrl)
.messageBody(theMessage)
.build());
// Handle the response asynchronously
future.whenComplete((response, throwable) -> {
if (throwable == null) {
producedCount.incrementAndGet();
} else if (!(throwable instanceof java.util.concurrent.CancellationException) &&
!(throwable.getMessage() != null && throwable.getMessage().contains("executor not accepting a task"))) {
log.error("BatchProducer: Failed to send message", throwable);
}
// Ignore CancellationException and executor shutdown errors - expected during shutdown
});
}
// Small delay to allow batching to occur
Thread.sleep(10);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("BatchProducer interrupted: " + e.getMessage());
} catch (SdkException | java.util.concurrent.CompletionException e) {
log.error("BatchProducer: " + e.getMessage());
System.exit(1);
}
}
}
/**
* Receives and deletes messages individually using SqsAsyncClient for baseline measurement.
*
* This consumer demonstrates traditional single-message operations without batching.
* Each receiveMessage() and deleteMessage() call results in separate API requests,
* providing a performance baseline for comparison with batch operations.
*
* The method signatures are identical to SqsAsyncBatchManager methods:
* - receiveMessage() matches batchManager.receiveMessage()
* - deleteMessage() matches batchManager.deleteMessage()
*
* This API consistency allows easy migration to the high-level batching library.
*/
private static class Consumer extends Thread {
final SqsAsyncClient sqsAsyncClient;
final String queueUrl;
final AtomicInteger consumedCount;
final AtomicBoolean stop;
/**
* Creates a consumer thread for single-message operations.
*
* @param sqsAsyncClient the SQS client for receiving messages
* @param queueUrl the URL of the source queue
* @param consumedCount shared counter for tracking processed messages
* @param stop shared flag to signal thread termination
*/
Consumer(SqsAsyncClient sqsAsyncClient, String queueUrl, AtomicInteger consumedCount,
AtomicBoolean stop) {
this.sqsAsyncClient = sqsAsyncClient;
this.queueUrl = queueUrl;
this.consumedCount = consumedCount;
this.stop = stop;
}
/**
* Continuously receives and deletes messages using traditional single-request operations.
*
* Uses SqsAsyncClient methods directly:
* - receiveMessage(): One API call per receive operation
* - deleteMessage(): One API call per delete operation
*
* This approach demonstrates the baseline performance without batching optimization.
* Compare these method calls with the identical signatures used in BatchConsumer
* to see how the high-level batching library maintains API compatibility.
*/
public void run() {
try {
while (!stop.get()) {
try {
final ReceiveMessageResponse result = sqsAsyncClient.receiveMessage(
ReceiveMessageRequest.builder()
.queueUrl(queueUrl)
.build()).join();
if (!result.messages().isEmpty()) {
final Message m = result.messages().get(0);
// Note: deleteMessage() signature identical to batchManager.deleteMessage()
sqsAsyncClient.deleteMessage(DeleteMessageRequest.builder()
.queueUrl(queueUrl)
.receiptHandle(m.receiptHandle())
.build()).join();
consumedCount.incrementAndGet();
}
} catch (SdkException | java.util.concurrent.CompletionException e) {
log.error(e.getMessage());
}
}
} catch (SdkException | java.util.concurrent.CompletionException e) {
// Handle both SdkException and CompletionException from async operations.
// If this unlikely condition occurs, stop.
log.error("Consumer: " + e.getMessage());
System.exit(1);
}
}
}
/**
* Receives and deletes messages using SqsAsyncBatchManager for automatic optimization.
*
* This consumer demonstrates the AWS SDK for Java 2.x high-level batching library
* for message consumption. The SqsAsyncBatchManager provides two key optimizations:
*
* 1. Receive optimization: Maintains an internal buffer of messages fetched in the
* background, so receiveMessage() calls return immediately from the buffer
* 2. Delete batching: Automatically buffers deleteMessage() calls and sends them
* as DeleteMessageBatchRequest operations when thresholds are reached
*
* Key features:
* - Identical API: receiveMessage() and deleteMessage() have the same signatures
* as SqsAsyncClient methods, making this a true drop-in replacement
* - Background fetching: The batch manager continuously fetches messages to keep
* the internal buffer populated, reducing receive latency
* - Automatic delete batching: Individual deleteMessage() calls are buffered and
* sent as batch operations (up to 10 per batch, 200ms frequency)
* - Transparent optimization: No application logic changes needed to benefit
*
* Performance benefits:
* - Reduced API calls through automatic batching of delete operations
* - Lower latency for receives due to background message buffering
* - Higher overall throughput with fewer network round trips
*/
private static class BatchConsumer extends Thread {
final SqsAsyncBatchManager batchManager;
final String queueUrl;
final int batchSize;
final AtomicInteger consumedCount;
final AtomicBoolean stop;
/**
* Creates a consumer thread for batch operations.
*
* @param batchManager the batch manager for efficient message processing
* @param queueUrl the URL of the source queue
* @param batchSize the maximum number of messages to receive per batch
* @param consumedCount shared counter for tracking processed messages
* @param stop shared flag to signal thread termination
*/
BatchConsumer(SqsAsyncBatchManager batchManager, String queueUrl, int batchSize,
AtomicInteger consumedCount, AtomicBoolean stop) {
this.batchManager = batchManager;
this.queueUrl = queueUrl;
this.batchSize = batchSize;
this.consumedCount = consumedCount;
this.stop = stop;
}
/**
* Continuously receives and deletes messages using the high-level batching library.
*
* Demonstrates the key advantage of SqsAsyncBatchManager: identical method signatures
* with automatic optimization. Notice how:
*
* - batchManager.receiveMessage() uses the same syntax as sqsAsyncClient.receiveMessage()
* - batchManager.deleteMessage() uses the same syntax as sqsAsyncClient.deleteMessage()
*
* Behind the scenes, the batch manager:
* 1. Maintains an internal message buffer populated by background fetching
* 2. Returns messages immediately from the buffer (reduced latency)
* 3. Automatically batches deleteMessage() calls into DeleteMessageBatchRequest operations
* 4. Sends batch deletes when maxBatchSize (10) or sendRequestFrequency (200ms) is reached
*
* This provides significant performance improvements with zero code changes
* compared to traditional SqsAsyncClient usage patterns.
*/
public void run() {
try {
while (!stop.get()) {
// Receive messages using the high-level batch manager.
// This call uses identical syntax to sqsAsyncClient.receiveMessage()
// but benefits from internal message buffering for improved performance.
final ReceiveMessageResponse result = batchManager.receiveMessage(
ReceiveMessageRequest.builder()
.queueUrl(queueUrl)
.maxNumberOfMessages(Math.min(batchSize, 10))
.build()).join();
if (!result.messages().isEmpty()) {
final List messages = result.messages();
// Delete messages using the batch manager.
// Each deleteMessage() call uses identical syntax to SqsAsyncClient
// but the SDK automatically buffers these calls and sends them
// as DeleteMessageBatchRequest operations for optimal performance.
for (Message message : messages) {
CompletableFuture future = batchManager.deleteMessage(
DeleteMessageRequest.builder()
.queueUrl(queueUrl)
.receiptHandle(message.receiptHandle())
.build());
future.whenComplete((response, throwable) -> {
if (throwable == null) {
consumedCount.incrementAndGet();
} else if (!(throwable instanceof java.util.concurrent.CancellationException) &&
!(throwable.getMessage() != null && throwable.getMessage().contains("executor not accepting a task"))) {
log.error("BatchConsumer: Failed to delete message", throwable);
}
// Ignore CancellationException and executor shutdown errors - expected during shutdown
});
}
}
// Small delay to prevent tight polling
Thread.sleep(10);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("BatchConsumer interrupted: " + e.getMessage());
} catch (SdkException | java.util.concurrent.CompletionException e) {
// Handle both SdkException and CompletionException from async operations.
// If this unlikely condition occurs, stop.
log.error("BatchConsumer: " + e.getMessage());
System.exit(1);
}
}
}
/**
* Displays real-time throughput statistics every second.
*
* This thread logs the current count of produced and consumed messages
* to help you monitor the performance comparison.
*/
private static class Monitor extends Thread {
private final AtomicInteger producedCount;
private final AtomicInteger consumedCount;
private final AtomicBoolean stop;
/**
* Creates a monitoring thread that displays throughput statistics.
*
* @param producedCount shared counter for messages sent
* @param consumedCount shared counter for messages processed
* @param stop shared flag to signal thread termination
*/
Monitor(AtomicInteger producedCount, AtomicInteger consumedCount,
AtomicBoolean stop) {
this.producedCount = producedCount;
this.consumedCount = consumedCount;
this.stop = stop;
}
/**
* Logs throughput statistics every second until stopped.
*
* Displays the current count of produced and consumed messages
* to help monitor the performance comparison between batching strategies.
*/
public void run() {
try {
while (!stop.get()) {
Thread.sleep(1000);
log.info("produced messages = " + producedCount.get()
+ ", consumed messages = " + consumedCount.get());
}
} catch (InterruptedException e) {
// Allow the thread to exit.
}
}
}
}
```
+ For API details, see the following topics in *AWS SDK for Java 2.x API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/CreateQueue)
+ [DeleteMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteMessage)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteQueue)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/ReceiveMessage)
+ [SendMessage](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/SendMessage)
+ [SendMessageBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/SendMessageBatch)
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples).
Create functions to wrap Amazon SQS message functions.
```
import logging
import sys
import boto3
from botocore.exceptions import ClientError
import queue_wrapper
logger = logging.getLogger(__name__)
sqs = boto3.resource("sqs")
def send_messages(queue, messages):
"""
Send a batch of messages in a single request to an SQS queue.
This request may return overall success even when some messages were not sent.
The caller must inspect the Successful and Failed lists in the response and
resend any failed messages.
:param queue: The queue to receive the messages.
:param messages: The messages to send to the queue. These are simplified to
contain only the message body and attributes.
:return: The response from SQS that contains the list of successful and failed
messages.
"""
try:
entries = [
{
"Id": str(ind),
"MessageBody": msg["body"],
"MessageAttributes": msg["attributes"],
}
for ind, msg in enumerate(messages)
]
response = queue.send_messages(Entries=entries)
if "Successful" in response:
for msg_meta in response["Successful"]:
logger.info(
"Message sent: %s: %s",
msg_meta["MessageId"],
messages[int(msg_meta["Id"])]["body"],
)
if "Failed" in response:
for msg_meta in response["Failed"]:
logger.warning(
"Failed to send: %s: %s",
msg_meta["MessageId"],
messages[int(msg_meta["Id"])]["body"],
)
except ClientError as error:
logger.exception("Send messages failed to queue: %s", queue)
raise error
else:
return response
def receive_messages(queue, max_number, wait_time):
"""
Receive a batch of messages in a single request from an SQS queue.
:param queue: The queue from which to receive messages.
:param max_number: The maximum number of messages to receive. The actual number
of messages received might be less.
:param wait_time: The maximum time to wait (in seconds) before returning. When
this number is greater than zero, long polling is used. This
can result in reduced costs and fewer false empty responses.
:return: The list of Message objects received. These each contain the body
of the message and metadata and custom attributes.
"""
try:
messages = queue.receive_messages(
MessageAttributeNames=["All"],
MaxNumberOfMessages=max_number,
WaitTimeSeconds=wait_time,
)
for msg in messages:
logger.info("Received message: %s: %s", msg.message_id, msg.body)
except ClientError as error:
logger.exception("Couldn't receive messages from queue: %s", queue)
raise error
else:
return messages
def delete_messages(queue, messages):
"""
Delete a batch of messages from a queue in a single request.
:param queue: The queue from which to delete the messages.
:param messages: The list of messages to delete.
:return: The response from SQS that contains the list of successful and failed
message deletions.
"""
try:
entries = [
{"Id": str(ind), "ReceiptHandle": msg.receipt_handle}
for ind, msg in enumerate(messages)
]
response = queue.delete_messages(Entries=entries)
if "Successful" in response:
for msg_meta in response["Successful"]:
logger.info("Deleted %s", messages[int(msg_meta["Id"])].receipt_handle)
if "Failed" in response:
for msg_meta in response["Failed"]:
logger.warning(
"Could not delete %s", messages[int(msg_meta["Id"])].receipt_handle
)
except ClientError:
logger.exception("Couldn't delete messages from queue %s", queue)
else:
return response
```
Use the wrapper functions to send and receive messages in batches.
```
def usage_demo():
"""
Shows how to:
* Read the lines from this Python file and send the lines in
batches of 10 as messages to a queue.
* Receive the messages in batches until the queue is empty.
* Reassemble the lines of the file and verify they match the original file.
"""
def pack_message(msg_path, msg_body, msg_line):
return {
"body": msg_body,
"attributes": {
"path": {"StringValue": msg_path, "DataType": "String"},
"line": {"StringValue": str(msg_line), "DataType": "String"},
},
}
def unpack_message(msg):
return (
msg.message_attributes["path"]["StringValue"],
msg.body,
int(msg.message_attributes["line"]["StringValue"]),
)
print("-" * 88)
print("Welcome to the Amazon Simple Queue Service (Amazon SQS) demo!")
print("-" * 88)
queue = queue_wrapper.create_queue("sqs-usage-demo-message-wrapper")
with open(__file__) as file:
lines = file.readlines()
line = 0
batch_size = 10
received_lines = [None] * len(lines)
print(f"Sending file lines in batches of {batch_size} as messages.")
while line < len(lines):
messages = [
pack_message(__file__, lines[index], index)
for index in range(line, min(line + batch_size, len(lines)))
]
line = line + batch_size
send_messages(queue, messages)
print(".", end="")
sys.stdout.flush()
print(f"Done. Sent {len(lines) - 1} messages.")
print(f"Receiving, handling, and deleting messages in batches of {batch_size}.")
more_messages = True
while more_messages:
received_messages = receive_messages(queue, batch_size, 2)
print(".", end="")
sys.stdout.flush()
for message in received_messages:
path, body, line = unpack_message(message)
received_lines[line] = body
if received_messages:
delete_messages(queue, received_messages)
else:
more_messages = False
print("Done.")
if all([lines[index] == received_lines[index] for index in range(len(lines))]):
print(f"Successfully reassembled all file lines!")
else:
print(f"Uh oh, some lines were missed!")
queue.delete()
print("Thanks for watching!")
print("-" * 88)
```
+ For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/CreateQueue)
+ [DeleteMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessage)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteQueue)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/ReceiveMessage)
+ [SendMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SendMessage)
+ [SendMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SendMessageBatch)
------
# Use the AWS Message Processing Framework for .NET to publish and receive Amazon SQS messages
The following code example shows how to create applications that publish and receive Amazon SQS messages using the AWS Message Processing Framework for .NET.
------
#### [ .NET ]
**SDK for .NET**
Provides a tutorial for the AWS Message Processing Framework for .NET. The tutorial creates a web application that allows the user to publish an Amazon SQS message and a command-line application that receives the message.
For complete source code and instructions on how to set up and run, see the [full tutorial](https://docs.aws.amazon.com/sdk-for-net/latest/developer-guide/msg-proc-fw-get-started.html) in the AWS SDK for .NET Developer Guide and the example on [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/cross-service/MessageProcessingFramework).
**Services used in this example**
+ Amazon SQS
------
# Use the Amazon SQS Java Messaging Library to work with the Java Message Service (JMS) interface for Amazon SQS
The following code example shows how to use the Amazon SQS Java Messaging Library to work with the JMS interface.
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sqs#code-examples).
The following examples work with standard Amazon SQS queues and include:
+ Sending a text message.
+ Receiving messages synchronously.
+ Receiving messages asynchronously.
+ Receiving messages using CLIENT\$1ACKNOWLEDGE mode.
+ Receiving messages using the UNORDERED\$1ACKNOWLEDGE mode.
+ Using Spring to inject dependencies.
+ A utility class that provides common methods used by the other examples.
For more information on using JMS with Amazon SQS, see the [Amazon SQS Developer Guide](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-java-message-service-jms-client.html).
Sending a text message.
```
/**
* This method establishes a connection to a standard Amazon SQS queue using the Amazon SQS
* Java Messaging Library and sends text messages to it. It uses JMS (Java Message Service) API
* with automatic acknowledgment mode to ensure reliable message delivery, and automatically
* manages all messaging resources.
*
* @throws JMSException If there is a problem connecting to or sending messages to the queue
*/
public static void doSendTextMessage() throws JMSException {
// Create a connection factory.
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
// Create the connection in a try-with-resources statement so that it's closed automatically.
try (SQSConnection connection = connectionFactory.createConnection()) {
// Create the queue if needed.
SqsJmsExampleUtils.ensureQueueExists(connection, QUEUE_NAME, SqsJmsExampleUtils.QUEUE_VISIBILITY_TIMEOUT);
// Create a session that uses the JMS auto-acknowledge mode.
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
createAndSendMessages(session, producer);
} // The connection closes automatically. This also closes the session.
LOGGER.info("Connection closed");
}
/**
* This method reads text input from the keyboard and sends each line as a separate message
* to a standard Amazon SQS queue using the Amazon SQS Java Messaging Library. It continues
* to accept input until the user enters an empty line, using JMS (Java Message Service) API to
* handle the message delivery.
*
* @param session The JMS session used to create messages
* @param producer The JMS message producer used to send messages to the queue
*/
private static void createAndSendMessages(Session session, MessageProducer producer) {
BufferedReader inputReader = new BufferedReader(
new InputStreamReader(System.in, Charset.defaultCharset()));
try {
String input;
while (true) {
LOGGER.info("Enter message to send (leave empty to exit): ");
input = inputReader.readLine();
if (input == null || input.isEmpty()) break;
TextMessage message = session.createTextMessage(input);
producer.send(message);
LOGGER.info("Send message {}", message.getJMSMessageID());
}
} catch (EOFException e) {
// Just return on EOF
} catch (IOException e) {
LOGGER.error("Failed reading input: {}", e.getMessage(), e);
} catch (JMSException e) {
LOGGER.error("Failed sending message: {}", e.getMessage(), e);
}
}
```
Receiving messages synchronously.
```
/**
* This method receives messages from a standard Amazon SQS queue using the Amazon SQS Java
* Messaging Library. It creates a connection to the queue using JMS (Java Message Service),
* waits for messages to arrive, and processes them one at a time. The method handles all
* necessary setup and cleanup of messaging resources.
*
* @throws JMSException If there is a problem connecting to or receiving messages from the queue
*/
public static void doReceiveMessageSync() throws JMSException {
// Create a connection factory.
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
// Create a connection.
try (SQSConnection connection = connectionFactory.createConnection() ) {
// Create the queue if needed.
SqsJmsExampleUtils.ensureQueueExists(connection, QUEUE_NAME, SqsJmsExampleUtils.QUEUE_VISIBILITY_TIMEOUT);
// Create a session.
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
connection.start();
receiveMessages(consumer);
} // The connection closes automatically. This also closes the session.
LOGGER.info("Connection closed");
}
/**
* This method continuously checks for new messages from a standard Amazon SQS queue using
* the Amazon SQS Java Messaging Library. It waits up to 20 seconds for each message, processes
* it using JMS (Java Message Service), and confirms receipt. The method stops checking for
* messages after 20 seconds of no activity.
*
* @param consumer The JMS message consumer that receives messages from the queue
*/
private static void receiveMessages(MessageConsumer consumer) {
try {
while (true) {
LOGGER.info("Waiting for messages...");
// Wait 1 minute for a message
Message message = consumer.receive(Duration.ofSeconds(20).toMillis());
if (message == null) {
LOGGER.info("Shutting down after 20 seconds of silence.");
break;
}
SqsJmsExampleUtils.handleMessage(message);
message.acknowledge();
LOGGER.info("Acknowledged message {}", message.getJMSMessageID());
}
} catch (JMSException e) {
LOGGER.error("Error receiving from SQS: {}", e.getMessage(), e);
}
}
```
Receiving messages asynchronously.
```
/**
* This method sets up automatic message handling for a standard Amazon SQS queue using the
* Amazon SQS Java Messaging Library. It creates a listener that processes messages as soon
* as they arrive using JMS (Java Message Service), runs for 5 seconds, then cleans up all
* messaging resources.
*
* @throws JMSException If there is a problem connecting to or receiving messages from the queue
*/
public static void doReceiveMessageAsync() throws JMSException {
// Create a connection factory.
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
// Create a connection.
try (SQSConnection connection = connectionFactory.createConnection() ) {
// Create the queue if needed.
SqsJmsExampleUtils.ensureQueueExists(connection, QUEUE_NAME, SqsJmsExampleUtils.QUEUE_VISIBILITY_TIMEOUT);
// Create a session.
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
try {
// Create a consumer for the queue.
MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
// Provide an implementation of the MessageListener interface, which has a single 'onMessage' method.
// We use a lambda expression for the implementation.
consumer.setMessageListener(message -> {
try {
SqsJmsExampleUtils.handleMessage(message);
message.acknowledge();
} catch (JMSException e) {
LOGGER.error("Error processing message: {}", e.getMessage());
}
});
// Start receiving incoming messages.
connection.start();
LOGGER.info("Waiting for messages...");
} catch (JMSException e) {
throw new RuntimeException(e);
}
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
} // The connection closes automatically. This also closes the session.
LOGGER.info( "Connection closed" );
}
```
Receiving messages using CLIENT\$1ACKNOWLEDGE mode.
```
/**
* This method demonstrates how message acknowledgment affects message processing in a standard
* Amazon SQS queue using the Amazon SQS Java Messaging Library. It sends messages to the queue,
* then shows how JMS (Java Message Service) client acknowledgment mode handles both explicit
* and implicit message confirmations, including how acknowledging one message can automatically
* acknowledge previous messages.
*
* @throws JMSException If there is a problem with the messaging operations
*/
public static void doReceiveMessagesSyncClientAcknowledge() throws JMSException {
// Create a connection factory.
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
// Create the connection in a try-with-resources statement so that it's closed automatically.
try (SQSConnection connection = connectionFactory.createConnection() ) {
// Create the queue if needed.
SqsJmsExampleUtils.ensureQueueExists(connection, QUEUE_NAME, TIME_OUT_SECONDS);
// Create a session with client acknowledge mode.
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
// Create a producer and consumer.
MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
// Open the connection.
connection.start();
// Send two text messages.
sendMessage(producer, session, "Message 1");
sendMessage(producer, session, "Message 2");
// Receive a message and don't acknowledge it.
receiveMessage(consumer, false);
// Receive another message and acknowledge it.
receiveMessage(consumer, true);
// Wait for the visibility time out, so that unacknowledged messages reappear in the queue,
LOGGER.info("Waiting for visibility timeout...");
try {
Thread.sleep(TIME_OUT_MILLIS);
} catch (InterruptedException e) {
LOGGER.error("Interrupted while waiting for visibility timeout", e);
Thread.currentThread().interrupt();
throw new RuntimeException("Processing interrupted", e);
}
/* We will attempt to receive another message, but none will be available. This is because in
CLIENT_ACKNOWLEDGE mode, when we acknowledged the second message, all previous messages were
automatically acknowledged as well. Therefore, although we never directly acknowledged the first
message, it was implicitly acknowledged when we confirmed the second one. */
receiveMessage(consumer, true);
} // The connection closes automatically. This also closes the session.
LOGGER.info("Connection closed.");
}
/**
* Sends a text message using the specified JMS MessageProducer and Session.
*
* @param producer The JMS MessageProducer used to send the message
* @param session The JMS Session used to create the text message
* @param messageText The text content to be sent in the message
* @throws JMSException If there is an error creating or sending the message
*/
private static void sendMessage(MessageProducer producer, Session session, String messageText) throws JMSException {
// Create a text message and send it.
producer.send(session.createTextMessage(messageText));
}
/**
* Receives and processes a message from a JMS queue using the specified consumer.
* The method waits for a message until the configured timeout period is reached.
* If a message is received, it is logged and optionally acknowledged based on the
* acknowledge parameter.
*
* @param consumer The JMS MessageConsumer used to receive messages from the queue
* @param acknowledge Boolean flag indicating whether to acknowledge the message.
* If true, the message will be acknowledged after processing
* @throws JMSException If there is an error receiving, processing, or acknowledging the message
*/
private static void receiveMessage(MessageConsumer consumer, boolean acknowledge) throws JMSException {
// Receive a message.
Message message = consumer.receive(TIME_OUT_MILLIS);
if (message == null) {
LOGGER.info("Queue is empty!");
} else {
// Since this queue has only text messages, cast the message object and print the text.
LOGGER.info("Received: {} Acknowledged: {}", ((TextMessage) message).getText(), acknowledge);
// Acknowledge the message if asked.
if (acknowledge) message.acknowledge();
}
}
```
Receiving messages using the UNORDERED\$1ACKNOWLEDGE mode.
```
/**
* Demonstrates message acknowledgment behavior in UNORDERED_ACKNOWLEDGE mode with Amazon SQS JMS.
* In this mode, each message must be explicitly acknowledged regardless of receive order.
* Unacknowledged messages return to the queue after the visibility timeout expires,
* unlike CLIENT_ACKNOWLEDGE mode where acknowledging one message acknowledges all previous messages.
*
* @throws JMSException If a JMS-related error occurs during message operations
*/
public static void doReceiveMessagesUnorderedAcknowledge() throws JMSException {
// Create a connection factory.
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
// Create the connection in a try-with-resources statement so that it's closed automatically.
try( SQSConnection connection = connectionFactory.createConnection() ) {
// Create the queue if needed.
SqsJmsExampleUtils.ensureQueueExists(connection, QUEUE_NAME, TIME_OUT_SECONDS);
// Create a session with unordered acknowledge mode.
Session session = connection.createSession(false, SQSSession.UNORDERED_ACKNOWLEDGE);
// Create the producer and consumer.
MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
// Open a connection.
connection.start();
// Send two text messages.
sendMessage(producer, session, "Message 1");
sendMessage(producer, session, "Message 2");
// Receive a message and don't acknowledge it.
receiveMessage(consumer, false);
// Receive another message and acknowledge it.
receiveMessage(consumer, true);
// Wait for the visibility time out, so that unacknowledged messages reappear in the queue.
LOGGER.info("Waiting for visibility timeout...");
try {
Thread.sleep(TIME_OUT_MILLIS);
} catch (InterruptedException e) {
LOGGER.error("Interrupted while waiting for visibility timeout", e);
Thread.currentThread().interrupt();
throw new RuntimeException("Processing interrupted", e);
}
/* We will attempt to receive another message, and we'll get the first message again. This occurs
because in UNORDERED_ACKNOWLEDGE mode, each message requires its own separate acknowledgment.
Since we only acknowledged the second message, the first message remains in the queue for
redelivery. */
receiveMessage(consumer, true);
LOGGER.info("Connection closed.");
} // The connection closes automatically. This also closes the session.
}
/**
* Sends a text message to an Amazon SQS queue using JMS.
*
* @param producer The JMS MessageProducer for the queue
* @param session The JMS Session for message creation
* @param messageText The message content
* @throws JMSException If message creation or sending fails
*/
private static void sendMessage(MessageProducer producer, Session session, String messageText) throws JMSException {
// Create a text message and send it.
producer.send(session.createTextMessage(messageText));
}
/**
* Synchronously receives a message from an Amazon SQS queue using the JMS API
* with an acknowledgment parameter.
*
* @param consumer The JMS MessageConsumer for the queue
* @param acknowledge If true, acknowledges the message after receipt
* @throws JMSException If message reception or acknowledgment fails
*/
private static void receiveMessage(MessageConsumer consumer, boolean acknowledge) throws JMSException {
// Receive a message.
Message message = consumer.receive(TIME_OUT_MILLIS);
if (message == null) {
LOGGER.info("Queue is empty!");
} else {
// Since this queue has only text messages, cast the message object and print the text.
LOGGER.info("Received: {} Acknowledged: {}", ((TextMessage) message).getText(), acknowledge);
// Acknowledge the message if asked.
if (acknowledge) message.acknowledge();
}
}
```
Using Spring to inject dependencies.
```
package com.example.sqs.jms.spring;
import com.amazon.sqs.javamessaging.SQSConnection;
import com.example.sqs.jms.SqsJmsExampleUtils;
import jakarta.jms.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.context.support.FileSystemXmlApplicationContext;
import java.io.File;
import java.net.URL;
import java.util.concurrent.TimeUnit;
/**
* Demonstrates how to send and receive messages using the Amazon SQS Java Messaging Library
* with Spring Framework integration. This example connects to a standard Amazon SQS message
* queue using Spring's dependency injection to configure the connection and messaging components.
* The application uses the JMS (Java Message Service) API to handle message operations.
*/
public class SpringExample {
private static final Integer POLLING_SECONDS = 15;
private static final String SPRING_XML_CONFIG_FILE = "SpringExampleConfiguration.xml.txt";
private static final Logger LOGGER = LoggerFactory.getLogger(SpringExample.class);
/**
* Demonstrates sending and receiving messages through a standard Amazon SQS message queue
* using Spring Framework configuration. This method loads connection settings from an XML file,
* establishes a messaging session using the Amazon SQS Java Messaging Library, and processes
* messages using JMS (Java Message Service) operations. If the queue doesn't exist, it will
* be created automatically.
*
* @param args Command line arguments (not used)
*/
public static void main(String[] args) {
URL resource = SpringExample.class.getClassLoader().getResource(SPRING_XML_CONFIG_FILE);
File springFile = new File(resource.getFile());
if (!springFile.exists() || !springFile.canRead()) {
LOGGER.error("File " + SPRING_XML_CONFIG_FILE + " doesn't exist or isn't readable.");
System.exit(1);
}
try (FileSystemXmlApplicationContext context =
new FileSystemXmlApplicationContext("file://" + springFile.getAbsolutePath())) {
Connection connection;
try {
connection = context.getBean(Connection.class);
} catch (NoSuchBeanDefinitionException e) {
LOGGER.error("Can't find the JMS connection to use: " + e.getMessage(), e);
System.exit(2);
return;
}
String queueName;
try {
queueName = context.getBean("queueName", String.class);
} catch (NoSuchBeanDefinitionException e) {
LOGGER.error("Can't find the name of the queue to use: " + e.getMessage(), e);
System.exit(3);
return;
}
try {
if (connection instanceof SQSConnection) {
SqsJmsExampleUtils.ensureQueueExists((SQSConnection) connection, queueName, SqsJmsExampleUtils.QUEUE_VISIBILITY_TIMEOUT);
}
// Create the JMS session.
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
SqsJmsExampleUtils.sendTextMessage(session, queueName);
MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
receiveMessages(consumer);
} catch (JMSException e) {
LOGGER.error(e.getMessage(), e);
throw new RuntimeException(e);
}
} // Spring context autocloses. Managed Spring beans that implement AutoClosable, such as the
// 'connection' bean, are also closed.
LOGGER.info("Context closed");
}
/**
* Continuously checks for and processes messages from a standard Amazon SQS message queue
* using the Amazon SQS Java Messaging Library underlying the JMS API. This method waits for incoming messages,
* processes them when they arrive, and acknowledges their receipt using JMS (Java Message
* Service) operations. The method will stop checking for messages after 15 seconds of
* inactivity.
*
* @param consumer The JMS message consumer used to receive messages from the queue
*/
private static void receiveMessages(MessageConsumer consumer) {
try {
while (true) {
LOGGER.info("Waiting for messages...");
// Wait 15 seconds for a message.
Message message = consumer.receive(TimeUnit.SECONDS.toMillis(POLLING_SECONDS));
if (message == null) {
LOGGER.info("Shutting down after {} seconds of silence.", POLLING_SECONDS);
break;
}
SqsJmsExampleUtils.handleMessage(message);
message.acknowledge();
LOGGER.info("Message acknowledged.");
}
} catch (JMSException e) {
LOGGER.error("Error receiving from SQS.", e);
}
}
}
```
Spring bean definitions.
```
```
A utility class that provides common methods used by the other examples.
```
package com.example.sqs.jms;
import com.amazon.sqs.javamessaging.AmazonSQSMessagingClientWrapper;
import com.amazon.sqs.javamessaging.ProviderConfiguration;
import com.amazon.sqs.javamessaging.SQSConnection;
import com.amazon.sqs.javamessaging.SQSConnectionFactory;
import jakarta.jms.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.exception.SdkException;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.CreateQueueRequest;
import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
import java.time.Duration;
import java.util.Base64;
import java.util.Map;
/**
* This utility class provides helper methods for working with Amazon Simple Queue Service (Amazon SQS)
* through the Java Message Service (JMS) interface. It contains common operations for managing message
* queues and handling message delivery.
*/
public class SqsJmsExampleUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(SqsJmsExampleUtils.class);
public static final Long QUEUE_VISIBILITY_TIMEOUT = 5L;
/**
* This method verifies that a message queue exists and creates it if necessary. The method checks for
* an existing queue first to optimize performance.
*
* @param connection The active connection to the messaging service
* @param queueName The name of the queue to verify or create
* @param visibilityTimeout The duration in seconds that messages will be hidden after being received
* @throws JMSException If there is an error accessing or creating the queue
*/
public static void ensureQueueExists(SQSConnection connection, String queueName, Long visibilityTimeout) throws JMSException {
AmazonSQSMessagingClientWrapper client = connection.getWrappedAmazonSQSClient();
/* In most cases, you can do this with just a 'createQueue' call, but 'getQueueUrl'
(called by 'queueExists') is a faster operation for the common case where the queue
already exists. Also, many users and roles have permission to call 'getQueueUrl'
but don't have permission to call 'createQueue'.
*/
if( !client.queueExists(queueName) ) {
CreateQueueRequest createQueueRequest = CreateQueueRequest.builder()
.queueName(queueName)
.attributes(Map.of(QueueAttributeName.VISIBILITY_TIMEOUT, String.valueOf(visibilityTimeout)))
.build();
client.createQueue( createQueueRequest );
}
}
/**
* This method sends a simple text message to a specified message queue. It handles all necessary
* setup for the message delivery process.
*
* @param session The active messaging session used to create and send the message
* @param queueName The name of the queue where the message will be sent
*/
public static void sendTextMessage(Session session, String queueName) {
// Rest of implementation...
try {
MessageProducer producer = session.createProducer( session.createQueue( queueName) );
Message message = session.createTextMessage("Hello world!");
producer.send(message);
} catch (JMSException e) {
LOGGER.error( "Error receiving from SQS", e );
}
}
/**
* This method processes incoming messages and logs their content based on the message type.
* It supports text messages, binary data, and Java objects.
*
* @param message The message to be processed and logged
* @throws JMSException If there is an error reading the message content
*/
public static void handleMessage(Message message) throws JMSException {
// Rest of implementation...
LOGGER.info( "Got message {}", message.getJMSMessageID() );
LOGGER.info( "Content: ");
if(message instanceof TextMessage txtMessage) {
LOGGER.info( "\t{}", txtMessage.getText() );
} else if(message instanceof BytesMessage byteMessage){
// Assume the length fits in an int - SQS only supports sizes up to 256k so that
// should be true
byte[] bytes = new byte[(int)byteMessage.getBodyLength()];
byteMessage.readBytes(bytes);
LOGGER.info( "\t{}", Base64.getEncoder().encodeToString( bytes ) );
} else if( message instanceof ObjectMessage) {
ObjectMessage objMessage = (ObjectMessage) message;
LOGGER.info( "\t{}", objMessage.getObject() );
}
}
/**
* This method sets up automatic message processing for a specified queue. It creates a listener
* that will receive and handle incoming messages without blocking the main program.
*
* @param session The active messaging session
* @param queueName The name of the queue to monitor
* @param connection The active connection to the messaging service
*/
public static void receiveMessagesAsync(Session session, String queueName, Connection connection) {
// Rest of implementation...
try {
// Create a consumer for the queue.
MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
// Provide an implementation of the MessageListener interface, which has a single 'onMessage' method.
// We use a lambda expression for the implementation.
consumer.setMessageListener(message -> {
try {
SqsJmsExampleUtils.handleMessage(message);
message.acknowledge();
} catch (JMSException e) {
LOGGER.error("Error processing message: {}", e.getMessage());
}
});
// Start receiving incoming messages.
connection.start();
} catch (JMSException e) {
throw new RuntimeException(e);
}
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
/**
* This method performs cleanup operations after message processing is complete. It receives
* any messages in the specified queue, removes the message queue and closes all
* active connections to prevent resource leaks.
*
* @param queueName The name of the queue to be removed
* @param visibilityTimeout The duration in seconds that messages are hidden after being received
* @throws JMSException If there is an error during the cleanup process
*/
public static void cleanUpExample(String queueName, Long visibilityTimeout) throws JMSException {
LOGGER.info("Performing cleanup.");
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
try (SQSConnection connection = connectionFactory.createConnection() ) {
ensureQueueExists(connection, queueName, visibilityTimeout);
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
receiveMessagesAsync(session, queueName, connection);
SqsClient sqsClient = connection.getWrappedAmazonSQSClient().getAmazonSQSClient();
try {
String queueUrl = sqsClient.getQueueUrl(b -> b.queueName(queueName)).queueUrl();
sqsClient.deleteQueue(b -> b.queueUrl(queueUrl));
LOGGER.info("Queue deleted: {}", queueUrl);
} catch (SdkException e) {
LOGGER.error("Error during SQS operations: ", e);
}
}
LOGGER.info("Clean up: Connection closed");
}
/**
* This method creates a background task that sends multiple messages to a specified queue
* after waiting for a set time period. The task operates independently to ensure efficient
* message processing without interrupting other operations.
*
* @param queueName The name of the queue where messages will be sent
* @param secondsToWait The number of seconds to wait before sending messages
* @param numMessages The number of messages to send
* @param visibilityTimeout The duration in seconds that messages remain hidden after being received
* @return A task that can be executed to send the messages
*/
public static Runnable sendAMessageAsync(String queueName, Long secondsToWait, Integer numMessages, Long visibilityTimeout) {
return () -> {
try {
Thread.sleep(Duration.ofSeconds(secondsToWait).toMillis());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
try {
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
SqsClient.create()
);
try (SQSConnection connection = connectionFactory.createConnection()) {
ensureQueueExists(connection, queueName, visibilityTimeout);
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
for (int i = 1; i <= numMessages; i++) {
MessageProducer producer = session.createProducer(session.createQueue(queueName));
producer.send(session.createTextMessage("Hello World " + i + "!"));
}
}
} catch (JMSException e) {
LOGGER.error(e.getMessage(), e);
throw new RuntimeException(e);
}
};
}
}
```
+ For API details, see the following topics in *AWS SDK for Java 2.x API Reference*.
+ [CreateQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/CreateQueue)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/DeleteQueue)
------
# Work with queue tags and Amazon SQS using an AWS SDK
The following code example shows how to perform tagging operation with Amazon SQS.
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/sqs#code-examples).
The following example creates tags for a queue, lists tags, and removes a tag.
```
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.ListQueueTagsResponse;
import software.amazon.awssdk.services.sqs.model.QueueDoesNotExistException;
import software.amazon.awssdk.services.sqs.model.SqsException;
import java.util.Map;
import java.util.UUID;
/**
* Before running this Java V2 code example, set up your development environment, including your credentials. For more
* information, see the AWS
* SDK for Java Developer Guide.
*/
public class TagExamples {
static final SqsClient sqsClient = SqsClient.create();
static final String queueName = "TagExamples-queue-" + UUID.randomUUID().toString().replace("-", "").substring(0, 20);
private static final Logger LOGGER = LoggerFactory.getLogger(TagExamples.class);
public static void main(String[] args) {
final String queueUrl;
try {
queueUrl = sqsClient.createQueue(b -> b.queueName(queueName)).queueUrl();
LOGGER.info("Queue created. The URL is: {}", queueUrl);
} catch (RuntimeException e) {
LOGGER.error("Program ending because queue was not created.");
throw new RuntimeException(e);
}
try {
addTags(queueUrl);
listTags(queueUrl);
removeTags(queueUrl);
} catch (RuntimeException e) {
LOGGER.error("Program ending because of an error in a method.");
} finally {
try {
sqsClient.deleteQueue(b -> b.queueUrl(queueUrl));
LOGGER.info("Queue successfully deleted. Program ending.");
sqsClient.close();
} catch (RuntimeException e) {
LOGGER.error("Program ending.");
} finally {
sqsClient.close();
}
}
}
/** This method demonstrates how to use a Java Map to a tag a aueue.
* @param queueUrl The URL of the queue to tag.
*/
public static void addTags(String queueUrl) {
// Build a map of the tags.
final Map tagsToAdd = Map.of(
"Team", "Development",
"Priority", "Beta",
"Accounting ID", "456def");
try {
// Add tags to the queue using a Consumer parameter.
sqsClient.tagQueue(b -> b
.queueUrl(queueUrl)
.tags(tagsToAdd)
);
} catch (QueueDoesNotExistException e) {
LOGGER.error("Queue does not exist: {}", e.getMessage(), e);
throw new RuntimeException(e);
}
}
/** This method demonstrates how to view the tags for a queue.
* @param queueUrl The URL of the queue whose tags you want to list.
*/
public static void listTags(String queueUrl) {
ListQueueTagsResponse response;
try {
// Call the listQueueTags method with a Consumer parameter that creates a ListQueueTagsRequest.
response = sqsClient.listQueueTags(b -> b
.queueUrl(queueUrl));
} catch (SqsException e) {
LOGGER.error("Exception thrown: {}", e.getMessage(), e);
throw new RuntimeException(e);
}
// Log the tags.
response.tags()
.forEach((k, v) ->
LOGGER.info("Key: {} -> Value: {}", k, v));
}
/**
* This method demonstrates how to remove tags from a queue.
* @param queueUrl The URL of the queue whose tags you want to remove.
*/
public static void removeTags(String queueUrl) {
try {
// Call the untagQueue method with a Consumer parameter.
sqsClient.untagQueue(b -> b
.queueUrl(queueUrl)
.tagKeys("Accounting ID") // Remove a single tag.
);
} catch (SqsException e) {
LOGGER.error("Exception thrown: {}", e.getMessage(), e);
throw new RuntimeException(e);
}
}
}
```
+ For API details, see the following topics in *AWS SDK for Java 2.x API Reference*.
+ [ListQueueTags](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/ListQueueTags)
+ [TagQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/TagQueue)
+ [UntagQueue](https://docs.aws.amazon.com/goto/SdkForJavaV2/sqs-2012-11-05/UntagQueue)
------
# Serverless examples for Amazon SQS
The following code examples show how to use Amazon SQS with AWS SDKs.
**Topics**
+ [Invoke a Lambda function from an Amazon SQS trigger](sqs_example_serverless_SQS_Lambda_section.md)
+ [Reporting batch item failures for Lambda functions with an Amazon SQS trigger](sqs_example_serverless_SQS_Lambda_batch_item_failures_section.md)
# Invoke a Lambda function from an Amazon SQS trigger
The following code examples show how to implement a Lambda function that receives an event triggered by receiving messages from an SQS queue. The function retrieves the messages from the event parameter and logs the content of each message.
------
#### [ .NET ]
**SDK for .NET**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using .NET.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace SqsIntegrationSampleCode
{
public async Task FunctionHandler(SQSEvent evnt, ILambdaContext context)
{
foreach (var message in evnt.Records)
{
await ProcessMessageAsync(message, context);
}
context.Logger.LogInformation("done");
}
private async Task ProcessMessageAsync(SQSEvent.SQSMessage message, ILambdaContext context)
{
try
{
context.Logger.LogInformation($"Processed message {message.Body}");
// TODO: Do interesting work based on the new message
await Task.CompletedTask;
}
catch (Exception e)
{
//You can use Dead Letter Queue to handle failures. By configuring a Lambda DLQ.
context.Logger.LogError($"An error occurred");
throw;
}
}
}
```
------
#### [ Go ]
**SDK for Go V2**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using Go.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package integration_sqs_to_lambda
import (
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(event events.SQSEvent) error {
for _, record := range event.Records {
err := processMessage(record)
if err != nil {
return err
}
}
fmt.Println("done")
return nil
}
func processMessage(record events.SQSMessage) error {
fmt.Printf("Processed message %s\n", record.Body)
// TODO: Do interesting work based on the new message
return nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using Java.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
public class Function implements RequestHandler {
@Override
public Void handleRequest(SQSEvent sqsEvent, Context context) {
for (SQSMessage msg : sqsEvent.getRecords()) {
processMessage(msg, context);
}
context.getLogger().log("done");
return null;
}
private void processMessage(SQSMessage msg, Context context) {
try {
context.getLogger().log("Processed message " + msg.getBody());
// TODO: Do interesting work based on the new message
} catch (Exception e) {
context.getLogger().log("An error occurred");
throw e;
}
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/blob/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using JavaScript.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const message of event.Records) {
await processMessageAsync(message);
}
console.info("done");
};
async function processMessageAsync(message) {
try {
console.log(`Processed message ${message.body}`);
// TODO: Do interesting work based on the new message
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
Consuming an SQS event with Lambda using TypeScript.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SQSEvent, Context, SQSHandler, SQSRecord } from "aws-lambda";
export const functionHandler: SQSHandler = async (
event: SQSEvent,
context: Context
): Promise => {
for (const message of event.Records) {
await processMessageAsync(message);
}
console.info("done");
};
async function processMessageAsync(message: SQSRecord): Promise {
try {
console.log(`Processed message ${message.body}`);
// TODO: Do interesting work based on the new message
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
------
#### [ PHP ]
**SDK for PHP**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using PHP.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws InvalidLambdaEvent
*/
public function handleSqs(SqsEvent $event, Context $context): void
{
foreach ($event->getRecords() as $record) {
$body = $record->getBody();
// TODO: Do interesting work based on the new message
}
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using Python.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
for message in event['Records']:
process_message(message)
print("done")
def process_message(message):
try:
print(f"Processed message {message['body']}")
# TODO: Do interesting work based on the new message
except Exception as err:
print("An error occurred")
raise err
```
------
#### [ Ruby ]
**SDK for Ruby**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using Ruby.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event:, context:)
event['Records'].each do |message|
process_message(message)
end
puts "done"
end
def process_message(message)
begin
puts "Processed message #{message['body']}"
# TODO: Do interesting work based on the new message
rescue StandardError => err
puts "An error occurred"
raise err
end
end
```
------
#### [ Rust ]
**SDK for Rust**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) repository.
Consuming an SQS event with Lambda using Rust.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::sqs::SqsEvent;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
event.payload.records.iter().for_each(|record| {
// process the record
tracing::info!("Message body: {}", record.body.as_deref().unwrap_or_default())
});
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
# Reporting batch item failures for Lambda functions with an Amazon SQS trigger
The following code examples show how to implement partial batch response for Lambda functions that receive events from an SQS queue. The function reports the batch item failures in the response, signaling to Lambda to retry those messages later.
------
#### [ .NET ]
**SDK for .NET**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using .NET.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace sqsSample;
public class Function
{
public async Task FunctionHandler(SQSEvent evnt, ILambdaContext context)
{
List batchItemFailures = new List();
foreach(var message in evnt.Records)
{
try
{
//process your message
await ProcessMessageAsync(message, context);
}
catch (System.Exception)
{
//Add failed message identifier to the batchItemFailures list
batchItemFailures.Add(new SQSBatchResponse.BatchItemFailure{ItemIdentifier=message.MessageId});
}
}
return new SQSBatchResponse(batchItemFailures);
}
private async Task ProcessMessageAsync(SQSEvent.SQSMessage message, ILambdaContext context)
{
if (String.IsNullOrEmpty(message.Body))
{
throw new Exception("No Body in SQS Message.");
}
context.Logger.LogInformation($"Processed message {message.Body}");
// TODO: Do interesting work based on the new message
await Task.CompletedTask;
}
}
```
------
#### [ Go ]
**SDK for Go V2**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using Go.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, sqsEvent events.SQSEvent) (map[string]interface{}, error) {
batchItemFailures := []map[string]interface{}{}
for _, message := range sqsEvent.Records {
if len(message.Body) > 0 {
// Your message processing condition here
fmt.Printf("Successfully processed message: %s\n", message.Body)
} else {
// Message processing failed
fmt.Printf("Failed to process message %s\n", message.MessageId)
batchItemFailures = append(batchItemFailures, map[string]interface{}{"itemIdentifier": message.MessageId})
}
}
sqsBatchResponse := map[string]interface{}{
"batchItemFailures": batchItemFailures,
}
return sqsBatchResponse, nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**SDK for Java 2.x**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using Java.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
import java.util.ArrayList;
import java.util.List;
public class ProcessSQSMessageBatch implements RequestHandler {
@Override
public SQSBatchResponse handleRequest(SQSEvent sqsEvent, Context context) {
List batchItemFailures = new ArrayList();
for (SQSEvent.SQSMessage message : sqsEvent.getRecords()) {
try {
//process your message
} catch (Exception e) {
//Add failed message identifier to the batchItemFailures list
batchItemFailures.add(new SQSBatchResponse.BatchItemFailure(message.getMessageId()));
}
}
return new SQSBatchResponse(batchItemFailures);
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using JavaScript.
```
// Node.js 20.x Lambda runtime, AWS SDK for Javascript V3
export const handler = async (event, context) => {
const batchItemFailures = [];
for (const record of event.Records) {
try {
await processMessageAsync(record, context);
} catch (error) {
batchItemFailures.push({ itemIdentifier: record.messageId });
}
}
return { batchItemFailures };
};
async function processMessageAsync(record, context) {
if (record.body && record.body.includes("error")) {
throw new Error("There is an error in the SQS Message.");
}
console.log(`Processed message: ${record.body}`);
}
```
Reporting SQS batch item failures with Lambda using TypeScript.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SQSEvent, SQSBatchResponse, Context, SQSBatchItemFailure, SQSRecord } from 'aws-lambda';
export const handler = async (event: SQSEvent, context: Context): Promise => {
const batchItemFailures: SQSBatchItemFailure[] = [];
for (const record of event.Records) {
try {
await processMessageAsync(record);
} catch (error) {
batchItemFailures.push({ itemIdentifier: record.messageId });
}
}
return {batchItemFailures: batchItemFailures};
};
async function processMessageAsync(record: SQSRecord): Promise {
if (record.body && record.body.includes("error")) {
throw new Error('There is an error in the SQS Message.');
}
console.log(`Processed message ${record.body}`);
}
```
------
#### [ PHP ]
**SDK for PHP**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using PHP.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handleSqs(SqsEvent $event, Context $context): void
{
$this->logger->info("Processing SQS records");
$records = $event->getRecords();
foreach ($records as $record) {
try {
// Assuming the SQS message is in JSON format
$message = json_decode($record->getBody(), true);
$this->logger->info(json_encode($message));
// TODO: Implement your custom processing logic here
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$this->markAsFailed($record);
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords SQS records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using Python.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
if event:
batch_item_failures = []
sqs_batch_response = {}
for record in event["Records"]:
try:
print(f"Processed message: {record['body']}")
except Exception as e:
batch_item_failures.append({"itemIdentifier": record['messageId']})
sqs_batch_response["batchItemFailures"] = batch_item_failures
return sqs_batch_response
```
------
#### [ Ruby ]
**SDK for Ruby**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda-with-batch-item-handling) repository.
Reporting SQS batch item failures with Lambda using Ruby.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'json'
def lambda_handler(event:, context:)
if event
batch_item_failures = []
sqs_batch_response = {}
event["Records"].each do |record|
begin
# process message
rescue StandardError => e
batch_item_failures << {"itemIdentifier" => record['messageId']}
end
end
sqs_batch_response["batchItemFailures"] = batch_item_failures
return sqs_batch_response
end
end
```
------
#### [ Rust ]
**SDK for Rust**
There's more on GitHub. Find the complete example and learn how to set up and run in the [Serverless examples](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) repository.
Reporting SQS batch item failures with Lambda using Rust.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::{
event::sqs::{SqsBatchResponse, SqsEvent},
sqs::{BatchItemFailure, SqsMessage},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn process_record(_: &SqsMessage) -> Result<(), Error> {
Err(Error::from("Error processing message"))
}
async fn function_handler(event: LambdaEvent) -> Result {
let mut batch_item_failures = Vec::new();
for record in event.payload.records {
match process_record(&record).await {
Ok(_) => (),
Err(_) => batch_item_failures.push(BatchItemFailure {
item_identifier: record.message_id.unwrap(),
}),
}
}
Ok(SqsBatchResponse {
batch_item_failures,
})
}
#[tokio::main]
async fn main() -> Result<(), Error> {
run(service_fn(function_handler)).await
}
```
------