AWS: Here's how you subscribe an SQS queue to an SNS topic!

Image via Wikipedia

There's precious few code examples for subscribing an Amazon Web Services' SQS queue to an SNS topic, especially using the SDK for Java. I've finally figured it out, and this is my way of giving back to the authors of all the hints I could find around the Web. The tricky part is the queue policy; if you create it incorrectly, you won't see any error messages -- it just won't work.

The methodology is this:

1. Create an SNS topic.
2. Set a policy on the topic to allow subscriptions.
3. Create an SQS queue.
4. Set a policy on the queue to allow message posting.
5. Subscribe the queue to the topic.
6. Wait for AWS settings to propagate and "take."
7. Send an SNS notification message.
8. Listen for the message on the queue.

And here's the code:


public class SnsWithSqsDemo {
private static final Log LOG = LogFactory.getLog(SnsWithSqsDemo.class);
private static final String TOPIC_NAME = "TestTopic";
private static final String QUEUE_NAME = "TestQueue";

/**
* This program demonstrates how to hook up an AWS SQS queue to an AWS SNS
* topic to receive messages.
*
* 1. Create an SNS topic.
* 2. Set a policy on the topic to allow subscriptions.
* 3. Create an SQS queue.
* 4. Set a policy on the queue to allow message posting.
* 5. Subscribe the queue to the topic.
* 6. Wait for AWS settings to propagate and "take."
* 7. Send an SNS notification message.
* 8. Listen for the message on the queue.
*
* @param args not used
*/
public static void main(String[] args) {
AWSCredentials awsCredentials = null;
AmazonSNS sns = null;
AmazonSQS sqs = null;
String topicArn = null;
String queueUrl = null;
String queueArn = null;
String subscriptionArn = null;
LOG.debug("Beginning.");
try {
awsCredentials =
new PropertiesCredentials(
SnsWithSqsDemo.class.getResourceAsStream("AwsCredentials.properties"));
sns = new AmazonSNSClient(awsCredentials);
// 1. Create a topic
System.out.println("Step 1. Create an SNS topic.");
CreateTopicResult ctResult =
sns.createTopic(new CreateTopicRequest(TOPIC_NAME));
topicArn = ctResult.getTopicArn();
System.out.println(String.format("Created topic %s with ARN %s",
TOPIC_NAME, topicArn));
// 2. Set policy on topic to allow open subscriptions
System.out.println("Step 2. Set a policy on the topic to allow subscriptions.");
Policy snsPolicy =
new Policy().withStatements(
new Statement(Effect.Allow)
.withPrincipals(Principal.AllUsers)
.withActions(SNSActions.Subscribe));
System.out.println("Set SNS policy: " + snsPolicy.toJson());
sns.setTopicAttributes(new SetTopicAttributesRequest(
topicArn, "Policy", snsPolicy.toJson()));
// 3. Create a queue
System.out.println("Step 3. Create an SQS queue.");
sqs = new AmazonSQSClient(awsCredentials);
CreateQueueResult cqResult =
sqs.createQueue(new CreateQueueRequest(QUEUE_NAME));
queueUrl = cqResult.getQueueUrl();
System.out.println(String.format("Created queue %s with URL %s",
QUEUE_NAME, queueUrl));
GetQueueAttributesResult queueArnResult =
sqs.getQueueAttributes(new GetQueueAttributesRequest(queueUrl)
.withAttributeNames("QueueArn"));
queueArn = queueArnResult.getAttributes().get("QueueArn");
System.out.println("Queue ARN = " + queueArn);
// 4. Set the queue policy to allow SNS to publish messages
System.out.println("Step 4. Set a policy on the queue to allow message posting.");
Policy sqsPolicy =
new Policy().withStatements(
new Statement(Effect.Allow)
.withPrincipals(Principal.AllUsers)
.withResources(new Resource(queueArn)) // Note: queue, not topic
.withActions(SQSActions.SendMessage)
.withConditions(
ConditionFactory.newSourceArnCondition(topicArn)));
Map queueAttributes = new HashMap();
queueAttributes.put("Policy", sqsPolicy.toJson());
sqs.setQueueAttributes(new SetQueueAttributesRequest(queueUrl, queueAttributes));
System.out.println("Set SQS policy to " + queueUrl + ": " + sqsPolicy.toJson());
// 5. Subscribe the queue to the topic
System.out.println("Step 5. Subscribe the queue to the topic.");
SubscribeResult sResult =
sns.subscribe(new SubscribeRequest(topicArn, "sqs", queueArn));
subscriptionArn = sResult.getSubscriptionArn();
System.out.println("Subscription ARN: " + subscriptionArn);
// 6. Wait a bit for AWS to get all synched-up
System.out.println("Step 6. Wait for AWS settings to propagate and \"take.\"");
Thread.sleep(60000L);
// 6.1. Verify queue attributes
GetQueueAttributesResult gqaResult =
sqs.getQueueAttributes(new GetQueueAttributesRequest(queueUrl)
.withAttributeNames("Policy", "QueueArn", "ApproximateNumberOfMessages"));
if (gqaResult.getAttributes().size() == 0) {
System.out.println("Queue " + QUEUE_NAME + " has no attributes");
} else {
System.out.println("Attributes for " + QUEUE_NAME);
for (String key : gqaResult.getAttributes().keySet()) {
System.out.println(String.format("\t%s = %s",
key, gqaResult.getAttributes().get(key)));
}
}
// 6.2. Verify topic attributes
GetTopicAttributesResult gtaResult =
sns.getTopicAttributes(new GetTopicAttributesRequest(topicArn));
if (gtaResult.getAttributes().size() == 0) {
System.out.println("Topic " + TOPIC_NAME + " has no attributes");
} else {
System.out.println("Attributes for " + TOPIC_NAME);
for (String key : gtaResult.getAttributes().keySet()) {
System.out.println(String.format("\t%s = %s",
key, gtaResult.getAttributes().get(key)));
}
}
// 6.3. Verify subscription
ListSubscriptionsByTopicResult lsbtResult =
sns.listSubscriptionsByTopic(new ListSubscriptionsByTopicRequest(topicArn));
if (lsbtResult.getSubscriptions().size() == 0) {
System.out.println("Topic " + TOPIC_NAME + " has no subscriptions.");
} else {
System.out.println("Subscriptions for " + TOPIC_NAME);
for (Subscription subscription : lsbtResult.getSubscriptions()) {
System.out.println("\t" + subscription.getProtocol() + ": "
+ subscription.getEndpoint());
}
}
// 7. Send a notification
System.out.println("Step 7. Send an SNS notification message.");
PublishResult pResult =
sns.publish(new PublishRequest(topicArn,
"Mr Watson -- Come here -- I want to see you."));
System.out.println("Sent message ID = " + pResult.getMessageId());
// 8. Wait for message receipt in queue
System.out.println("Step 8. Listen for the message on the queue.");
for (int i = 0; i < 10; i++) { Thread.sleep(2000L); ReceiveMessageResult rmResult = sqs.receiveMessage(new ReceiveMessageRequest(queueUrl)); if (rmResult.getMessages().size() > 0) {
// A message has been received
for (Message message : rmResult.getMessages()) {
System.out.println(message.getBody());
sqs.deleteMessage(new DeleteMessageRequest(queueUrl,
message.getReceiptHandle()));
}
break;
} else {
// ??? Why aren't we receiving messages?
System.out.println("No messages available, attempt " + (i+1));
}
}
} catch (Exception e) {
e.printStackTrace(System.err);
} finally {
System.out.println("Shutting down...");
// Unsubscribe the queue from the topic
if (sns != null && subscriptionArn != null) {
sns.unsubscribe(new UnsubscribeRequest(subscriptionArn));
System.out.println("Unsubscribed queue from topic.");
}
// Destroy queue
if (sqs != null && queueUrl != null) {
sqs.deleteQueue(new DeleteQueueRequest(queueUrl));
System.out.println("Deleted the queue.");
sqs.shutdown();
}
// Destroy topic
if (sns != null && topicArn != null) {
sns.deleteTopic(new DeleteTopicRequest(topicArn));
System.out.println("Deleted the topic.");
sns.shutdown();
}
}
LOG.debug("Done.");
}

}


AWS: How do you subscribe an SQS queue to an SNS topic?

Image via CrunchBase

If there are any AWS (Amazon Web Services) gurus out there, I'm stumped on trying to use a Simple Queue Service queue to receive notifications sent out on a Simple Notification Service topic. Any pointers?

In the code below, I'm using the AWS SDK for Java (available here). The program does the following:

1. Create an SNS topic.
2. Set a policy on the topic to allow subscriptions.
3. Create an SQS queue.
4. Set a policy on the queue to allow message posting.
5. Subscribe the queue to the topic.
6. Send an SNS notification message.
7. Listen for the message on the queue.

Steps 1-6 seem to work just fine, but on step 7, it listens in vain for messages. All ten iterations go by without picking up any messages on the queue. As an added note, if I subscribe an email address to the topic, messages are delivered. Only the queue appears to be bypassed when SNS sends its notifications.


public class SnsWithSqsDemo {
private static final String TOPIC_NAME = "TestTopic";
private static final String QUEUE_NAME = "TestQueue";

/**
* This program demonstrates how to hook up an AWS SQS queue to an AWS SNS
* topic to receive messages.
*
* 1. Create an SNS topic.
* 2. Set a policy on the topic to allow subscriptions.
* 3. Create an SQS queue.
* 4. Set a policy on the queue to allow message posting.
* 5. Subscribe the queue to the topic.
* 6. Send an SNS notification message.
* 7. Listen for the message on the queue.
*
* @param args not used
*/
public static void main(String[] args) {
AWSCredentials awsCredentials = null;
AmazonSNS sns = null;
AmazonSQS sqs = null;
String topicArn = null;
String queueUrl = null;
String queueArn = null;
String subscriptionArn = null;
try {
awsCredentials =
new PropertiesCredentials(
SnsWithSqsDemo.class.getResourceAsStream("AwsCredentials.properties"));
sns = new AmazonSNSClient(awsCredentials);
// 1. Create a topic
CreateTopicResult ctResult =
sns.createTopic(new CreateTopicRequest(TOPIC_NAME));
topicArn = ctResult.getTopicArn();
System.out.println(String.format("Created topic %s with ARN %s",
TOPIC_NAME, topicArn));
// 2. Set policy on topic to allow open subscriptions
Policy snsPolicy =
new Policy().withStatements(
new Statement(Effect.Allow)
.withPrincipals(Principal.AllUsers)
.withActions(SNSActions.Subscribe));
// ??? Is there no Java SDK constant for "Policy"?
sns.setTopicAttributes(new SetTopicAttributesRequest(
topicArn, "Policy", snsPolicy.toJson()));
// 3. Create a queue
sqs = new AmazonSQSClient(awsCredentials);
CreateQueueResult cqResult =
sqs.createQueue(new CreateQueueRequest(QUEUE_NAME));
queueUrl = cqResult.getQueueUrl();
System.out.println(String.format("Created queue %s with URL %s",
QUEUE_NAME, queueUrl));
// 4. Set the queue policy to allow SNS to publish messages
Policy sqsPolicy =
new Policy().withStatements(
new Statement(Effect.Allow)
.withPrincipals(Principal.AllUsers)
.withActions(SQSActions.SendMessage)
.withConditions(ConditionFactory.newSourceArnCondition(topicArn)));
Map queueAttributes = new HashMap();
queueAttributes.put("Policy", sqsPolicy.toJson());
sqs.setQueueAttributes(new SetQueueAttributesRequest(queueUrl, queueAttributes));
// ??? Why isn't the queue ARN in the queue attributes?
AmazonIdentityManagement idMgr = new AmazonIdentityManagementClient(awsCredentials);
String accountId = idMgr.getUser().getUser().getUserId();
queueArn = String.format("arn:aws:sqs:us-east-1:%s:%s", accountId, QUEUE_NAME);
// 5. Subscribe the queue to the topic
SubscribeResult sResult =
sns.subscribe(new SubscribeRequest(topicArn, "sqs", queueArn));
subscriptionArn = sResult.getSubscriptionArn();
System.out.println("Subscription ARN: " + subscriptionArn);
// 5.5. Wait a bit for AWS to get all synched-up
Thread.sleep(60000L);
// 6. Send a notification
PublishResult pResult =
sns.publish(new PublishRequest(topicArn,
"Mr Watson -- Come here -- I want to see you."));
System.out.println("Sent message ID = " + pResult.getMessageId());
// 7. Wait for message receipt in queue
for (int i = 0; i < 10; i++) { Thread.sleep(2000L); ReceiveMessageResult rmResult = sqs.receiveMessage(new ReceiveMessageRequest(queueUrl)); if (rmResult.getMessages().size() > 0) {
// A message has been received
for (Message message : rmResult.getMessages()) {
System.out.println(message.getBody());
sqs.deleteMessage(new DeleteMessageRequest(queueUrl,
message.getReceiptHandle()));
}
break;
} else {
System.out.println("No messages available, attempt " + (i+1));
}
}
} catch (Exception e) {
e.printStackTrace(System.err);
} finally {
// Unsubscribe the queue from the topic
if (sns != null && subscriptionArn != null) {
sns.unsubscribe(new UnsubscribeRequest(subscriptionArn));
System.out.println("Unsubscribed queue from topic.");
}
// Destroy queue
if (sqs != null && queueUrl != null) {
sqs.deleteQueue(new DeleteQueueRequest(queueUrl));
System.out.println("Deleted the queue.");
sqs.shutdown();
}
// Destroy topic
if (sns != null && topicArn != null) {
sns.deleteTopic(new DeleteTopicRequest(topicArn));
System.out.println("Deleted the topic.");
sns.shutdown();
}
}
}

}