&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();
}
```
+ API の詳細については、「*AWS SDK for C\$1\$1 API リファレンス*」の以下のトピックを参照してください。
+ [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**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/gov2/workflows/topics_and_queues#code-examples)での設定と実行の方法を確認してください。
コマンドプロンプトからインタラクティブなシナリオを実行します。
```
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))
}
```
この例で使用されている Amazon SNS アクションをラップする構造体を定義します。
```
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
}
```
この例で使用されている Amazon SQS アクションをラップする構造体を定義します。
```
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
}
```
リソースをクリーンアップします。
```
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)
}
}
}
```
+ API の詳細については、「*AWS SDK for Go API リファレンス*」の以下のトピックを参照してください。
+ [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**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](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 "";
}
}
```
+ API の詳細については、「*AWS SDK for Java 2.x API リファレンス*」の以下のトピックを参照してください。
+ [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)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/cross-services/wkflw-topics-queues#code-examples)での設定と実行の方法を確認してください。
これがこのシナリオのエントリーポイントです。
```
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();
};
```
上記のコードにより必要な依存関係が提供され、シナリオが開始されます。次のセクションには、この例の大部分が含まれています。
```
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();
}
}
}
```
+ API の詳細については、「*AWS SDK for JavaScript API リファレンス*」の以下のトピックを参照してください。
+ [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**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](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
}
}
```
+ API の詳細については、『*AWS SDK for Kotlin API リファレンス*』の以下のトピックを参照してください。
+ [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)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/topics_and_queues#code-examples)での設定と実行の方法を確認してください。
コマンドプロンプトからインタラクティブなシナリオを実行します。
```
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.")
```
シナリオで使用する Amazon SNS および Amazon SQS オペレーションをラップするクラスを作成します。
```
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
```
+ API の詳細については、「**AWS SDK for Python (Boto3) API リファレンス」の以下のトピックを参照してください。
+ [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**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](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)
}
}
}
```
+ API の詳細については、*AWS SDK for Swift API リファレンス*の以下のトピックを参照してください。
+ [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:))
------
# API Gateway を使用して Lambda 関数を呼び出す
次のコード例は、Amazon API Gateway によって呼び出される AWS Lambda 関数を作成する方法を示しています。
------
#### [ Java ]
**SDK for Java 2.x**
Lambda Java ランタイム API を使用して AWS Lambda 関数を作成する方法を示します。この例では、さまざまな AWS サービスを呼び出して、特定のユースケースを実行します。この例では、Amazon API Gateway によって呼び出される Lambda 関数を作成する方法を示します。この関数は、Amazon DynamoDB テーブルをスキャンして、Amazon Simple Notification Service (Amazon SNS) を使用して、従業員に年間の記念日を祝福するテキストメッセージを送信します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/usecases/creating_lambda_apigateway) で完全な例を参照してください。
**この例で使用されているサービス**
+ API ゲートウェイ
+ DynamoDB
+ Lambda
+ Amazon SNS
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
Lambda JavaScript ランタイム API を使用して AWS Lambda 関数を作成する方法を示します。この例では、さまざまな AWS サービスを呼び出して、特定のユースケースを実行します。この例では、Amazon API Gateway によって呼び出される Lambda 関数を作成する方法を示します。この関数は、Amazon DynamoDB テーブルをスキャンして、Amazon Simple Notification Service (Amazon SNS) を使用して、従業員に年間の記念日を祝福するテキストメッセージを送信します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/cross-services/lambda-api-gateway) で完全な例を参照してください。
この例は、[AWS SDK for JavaScript v3 デベロッパーガイド](https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/api-gateway-invoking-lambda-example.html)でも使用できます。
**この例で使用されているサービス**
+ API ゲートウェイ
+ DynamoDB
+ Lambda
+ Amazon SNS
------
#### [ Python ]
**SDK for Python (Boto3)**
この例は、 AWS Lambda 関数を対象とする Amazon API Gateway REST API を作成して使用する方法を示しています。Lambda ハンドラーは、HTTP メソッドに基づいてルーティングする方法を示します。クエリ文字列、ヘッダー、および本文からデータを取得する方法。そして、JSON 応答を返す方法。
+ Lambda 関数をデプロイします。
+ API ゲートウェイ REST API を作成します。
+ Lambda 関数をターゲットとする REST リソースを作成します。
+ API Gateway に Lambda 関数を呼び出す権限を付与します。
+ リクエストパッケージを使用して、REST API にリクエストを送信します。
+ デモ中に作成されたすべてのリソースをクリーンアップします。
この例は GitHub で最もよく確認できます。完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#readme) で完全な例を参照してください。
**この例で使用されているサービス**
+ API ゲートウェイ
+ DynamoDB
+ Lambda
+ Amazon SNS
------
# スケジュールされたイベントを使用した Lambda 関数の呼び出し
次のコード例は、Amazon EventBridge スケジュールされたイベントによって呼び出される AWS Lambda 関数を作成する方法を示しています。
------
#### [ Java ]
**SDK for Java 2.x**
AWS Lambda 関数を呼び出す Amazon EventBridge スケジュールされたイベントを作成する方法を示します。cron 式を使用して Lambda 関数が呼び出されるタイミングをスケジュールするように EventBridge を設定します。この例では、Lambda Java ランタイム API を使用して Lambda 関数を作成します。この例では、さまざまな AWS サービスを呼び出して、特定のユースケースを実行します。この例では、年間の記念日に従業員を祝福するモバイルテキストメッセージを従業員に送信するアプリを作成する方法を示します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/usecases/creating_scheduled_events) で完全な例を参照してください。
**この例で使用されているサービス**
+ CloudWatch Logs
+ DynamoDB
+ EventBridge
+ Lambda
+ Amazon SNS
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
AWS Lambda 関数を呼び出す Amazon EventBridge スケジュールされたイベントを作成する方法を示します。cron 式を使用して Lambda 関数が呼び出されるタイミングをスケジュールするように EventBridge を設定します。この例では、Lambda JavaScript ランタイム API を使用して Lambda 関数を作成します。この例では、さまざまな AWS サービスを呼び出して、特定のユースケースを実行します。この例では、年間の記念日に従業員を祝福するモバイルテキストメッセージを従業員に送信するアプリを作成する方法を示します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/cross-services/lambda-scheduled-events) で完全な例を参照してください。
この例は、[AWS SDK for JavaScript v3 デベロッパーガイド](https://docs.aws.amazon.com/sdk-for-javascript/v3/developer-guide/scheduled-events-invoking-lambda-example.html)でも使用できます。
**この例で使用されているサービス**
+ CloudWatch Logs
+ DynamoDB
+ EventBridge
+ Lambda
+ Amazon SNS
------
#### [ Python ]
**SDK for Python (Boto3)**
この例では、スケジュールされた Amazon EventBridge イベントのターゲットとして AWS Lambda 関数を登録する方法を示します。Lambda ハンドラーは、後で取得するために Amazon CloudWatch Logs にわかりやすいメッセージと完全なイベントデータを書き込みます。
+ Lambda 関数をデプロイします。
+ EventBridge スケジュールイベントを作成し、Lambda 関数をターゲットにします。
+ EventBridge に Lambda 関数を呼び出す許可を付与します
+ CloudWatch Logs から最新のデータを出力して、スケジュールされた呼び出しの結果を表示しています。
+ デモ中に作成されたすべてのリソースをクリーンアップします。
この例は GitHub で最もよく確認できます。完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#readme) で完全な例を参照してください。
**この例で使用されているサービス**
+ CloudWatch Logs
+ DynamoDB
+ EventBridge
+ Lambda
+ Amazon SNS
------
# Amazon SNS のサーバーレス例
次のコード例は、 AWS SDKs で Amazon SNS を使用する方法を示しています。
**Topics**
+ [Amazon SNS トリガーから Lambda 関数を呼び出す](sns_example_serverless_SNS_Lambda_section.md)
# Amazon SNS トリガーから Lambda 関数を呼び出す
次のコード例は、SNS トピックからメッセージを受信することによってトリガーされるイベントを受け取る Lambda 関数を実装する方法を示しています。この関数はイベントパラメータからメッセージを取得し、各メッセージの内容を記録します。
------
#### [ .NET ]
**SDK for .NET**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を検索し、設定および実行する方法を確認してください。
.NET を使用した Lambda での SNS イベントの消費。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SNSEvents;
// 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 SnsIntegration;
public class Function
{
public async Task FunctionHandler(SNSEvent evnt, ILambdaContext context)
{
foreach (var record in evnt.Records)
{
await ProcessRecordAsync(record, context);
}
context.Logger.LogInformation("done");
}
private async Task ProcessRecordAsync(SNSEvent.SNSRecord record, ILambdaContext context)
{
try
{
context.Logger.LogInformation($"Processed record {record.Sns.Message}");
// 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**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Go を使用して Lambda で SNS イベントを消費します。
```
// 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, snsEvent events.SNSEvent) {
for _, record := range snsEvent.Records {
processMessage(record)
}
fmt.Println("done")
}
func processMessage(record events.SNSEventRecord) {
message := record.SNS.Message
fmt.Printf("Processed message: %s\n", message)
// TODO: Process your record here
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**SDK for Java 2.x**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Java を使用した Lambda での SNS イベントの消費。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package example;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SNSEvent;
import com.amazonaws.services.lambda.runtime.events.SNSEvent.SNSRecord;
import java.util.Iterator;
import java.util.List;
public class SNSEventHandler implements RequestHandler {
LambdaLogger logger;
@Override
public Boolean handleRequest(SNSEvent event, Context context) {
logger = context.getLogger();
List records = event.getRecords();
if (!records.isEmpty()) {
Iterator recordsIter = records.iterator();
while (recordsIter.hasNext()) {
processRecord(recordsIter.next());
}
}
return Boolean.TRUE;
}
public void processRecord(SNSRecord record) {
try {
String message = record.getSNS().getMessage();
logger.log("message: " + message);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/blob/main/integration-sns-to-lambda)リポジトリで完全な例を検索し、設定および実行する方法を確認してください。
JavaScript を使用した Lambda での SNS イベントの消費。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
await processMessageAsync(record);
}
console.info("done");
};
async function processMessageAsync(record) {
try {
const message = JSON.stringify(record.Sns.Message);
console.log(`Processed message ${message}`);
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
TypeScript を使用した Lambda での SNS イベントの消費。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SNSEvent, Context, SNSHandler, SNSEventRecord } from "aws-lambda";
export const functionHandler: SNSHandler = async (
event: SNSEvent,
context: Context
): Promise => {
for (const record of event.Records) {
await processMessageAsync(record);
}
console.info("done");
};
async function processMessageAsync(record: SNSEventRecord): Promise {
try {
const message: string = JSON.stringify(record.Sns.Message);
console.log(`Processed message ${message}`);
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
------
#### [ PHP ]
**SDK for PHP**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
PHP を使用して Lambda で SNS イベントを消費します。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
getRecords() as $record) {
$message = $record->getMessage();
// TODO: Implement your custom processing logic here
// Any exception thrown will be logged and the invocation will be marked as failed
echo "Processed Message: $message" . PHP_EOL;
}
}
}
return new Handler();
```
------
#### [ Python ]
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用して Lambda で SNS イベントを消費します。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
for record in event['Records']:
process_message(record)
print("done")
def process_message(record):
try:
message = record['Sns']['Message']
print(f"Processed message {message}")
# TODO; Process your record here
except Exception as e:
print("An error occurred")
raise e
```
------
#### [ Ruby ]
**SDK for Ruby**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Ruby を使用した Lambda での SNS イベントの消費。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event:, context:)
event['Records'].map { |record| process_message(record) }
end
def process_message(record)
message = record['Sns']['Message']
puts("Processing message: #{message}")
rescue StandardError => e
puts("Error processing message: #{e}")
raise
end
```
------
#### [ Rust ]
**SDK for Rust**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Rust を使用して Lambda で SNS イベントを消費します。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::sns::SnsEvent;
use aws_lambda_events::sns::SnsRecord;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
use tracing::info;
// Built with the following dependencies:
// aws_lambda_events = { version = "0.10.0", default-features = false, features = ["sns"] }
// lambda_runtime = "0.8.1"
// tokio = { version = "1", features = ["macros"] }
// tracing = { version = "0.1", features = ["log"] }
// tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
for event in event.payload.records {
process_record(&event)?;
}
Ok(())
}
fn process_record(record: &SnsRecord) -> Result<(), Error> {
info!("Processing SNS Message: {}", record.sns.message);
// Implement your record handling code here.
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_target(false)
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------