Nav

Amazon SQS Connector

Introduction

Amazon Simple Queue Service (Amazon SQS) offers a reliable, highly scalable hosted queue for storing messages as they travel between computers. By using Amazon SQS, developers can simply move data between distributed application components performing different tasks, without losing messages or requiring each component to be always available. Amazon SQS makes it easy to build an automated workflow, working in close conjunction with the Amazon Elastic Compute Cloud (Amazon EC2) and the other AWS infrastructure web services.

The AWS SDK for Java provides a Java API for AWS infrastructure services. The Amazon SQS connector is built using the SDK for Java. For the complete list of operations supported by the connector, see Apidocs and Samples on github.io.

Although it is called a queue, Amazon SQS does not guarantee FIFO access to messages in Amazon SQS queues, mainly because of its distributed nature. If you require specific message ordering, you need to design your application to handle it.

Review the connector operations and functionality using the technical reference and demos.

MuleSoft maintains this connector under the Select support policy.

Prerequisites

This document assumes you are familiar with Mule, Anypoint Connectors, and Anypoint Studio Essentials. To increase your familiarity with Studio, consider completing one or more Anypoint Connector quickstart. Further, this page assumes that you have a basic understanding of Mule flows and Mule Global Elements.

This document describes implementation examples within the context of Anypoint Studio, Mule’s graphical user interface, and, in parallel, includes configuration details for doing the same in the XML Editor.

As the connector comes pre-bundled with Anypoint Studio, you must:

  • Install Anypoint Studio.

  • Sign Up for Amazon Web Services. To access AWS with the connector, you need the credentials in the form of IAM.

Compatibility

Amazon SQS connector is compatible with:

Application/Service Version

Mule Runtime

3.5.x or later

AWS SDK for Java

1.11.21

The package names of the arguments and the return types of most of the operations  have been updated in this version of the connector. Refer to the updated package names depending on how you are using the connector. For example, the Message Attributes input argument in the Send Message operation now expects a Map<String, MessageAttributeValue> object, which requires you to refer to the org.mule.modules.sqs.model.MessageAttributeValue package instead of com.amazonaws.services.sqs.model.MessageAttributeValue.

For all the operations in the connector to work, you need to update the subset of the overall list of Amazon SQS actions in the current SQS Policy with the new actions to specify which AWS account has access to the actions on a queue.

Ensure that sqs:GetQueueAttributes action is enabled for the queue under test in Amazon SQS policy as it is used by the Test Connection feature in the Global Configuration.

Installing the Connector

You can install a connector in Anypoint Studio using the instructions in Installing a Connector from Anypoint Exchange.

Updating from an Older Version

When a new version of a connector is released, Anypoint studio displays a popup in the bottom right corner of you screen with the following message: Updates Available. To upgrade to the newer version of the Amazon SQS connector:

  1. Click the popup and check for the available updates.

  2. Select the Amazon SQS connector 3.0 check-box and click Next.

  3. Follow the instructions provided by the user interface.

  4. Restart Studio when prompted.

  5. After restarting, if you have several versions of the connector installed, Mule asks you for the version of the connector you like to use.

Configuring an Amazon SQS Connector Global Element

To use the Amazon SQS connector in your Mule application, you must configure a global element that can be used by all the Amazon SQS connectors in the application (read more about global elements).

To create a global Amazon SQS connector configuration:

  1. Click the Global Elements tab at the base of the canvas.

  2. In the Global Mule Configuration Elements screen, click Create.

  3. In the Choose Global Type wizard, expand Connector Configuration, and then select Amazon SQS: Configuration.

    GlobalType

  4. Click OK.

  5. Enter the global element properties.

    Field Description

    Access Key

    Alphanumeric text string that uniquely identifies the user who owns the account. Required

    Secret Key

    Key that plays the role of a password. Required

    Try Default AWS Credentials Provider Chain

    Checkbox that controls whether temporary credentials should be used.

    Queue Name

    The default queue name; if it doesn’t exist, Mule automatically creates the queue.

    Queue URL

    The URL of the Amazon SQS queue to act upon.

    Region Endpoint

    The regional endpoint to process your requests.

    When a Queue Name is provided in the global element, the connector automatically creates the queue and sets the URL of this queue as Queue URL. All the Amazon SQS Message processors that reference the global element perform operations using this Queue URL.

    If you have to reference a different Queue URL for a particular message processor in the flow, you can perform the operation using the Queue URL attribute provided by the message processor.

  6. Keep the Pooling Profile and the Reconnection tabs with their default entries.

  7. Click Test Connection to confirm that the parameters of your global configuration are accurate, and that Mule is able to successfully connect to your instance of Amazon SQS. Read more about Testing Connections.

  8. Click OK to save the global connector configurations.

Using the Connector

The Amazon SQS connector is an operation-based connector, which means that when you add the connector to your flow, you need to configure a specific operation the connector is intended to perform.

See the technical reference documentation for the connector configurations and operations here.

The Amazon SQS connector supports the following operations:

  • Add Permission

  • Change message visibility

  • Change message visibility batch

  • Create queue

  • Delete message

  • Delete message batch

  • Delete queue

  • Get approximate number of messages

  • Get queue attributes

  • Get queue URL

  • List dead letter source queues

  • List queues

  • Purge Queue

  • Receive Messages

  • Remove permission

  • Send message batch

  • Send message

  • Set Queue Attributes

Connector Namespace and Schema

When designing your application in Studio, the act of dragging the connector from the palette onto the Anypoint Studio canvas should automatically populate the XML code with the connector namespace and schema location.

Namespace: http://www.mulesoft.org/schema/mule/sqs

Schema Location: http://www.mulesoft.org/schema/mule/sqs/current/mule-sqs.xsd

If you are manually coding the Mule application in Studio’s XML editor or other text editor, paste these into the header of your Configuration XML, inside the <mule> tag.

          
       
1
2
3
4
5
6
7
8
9
10
11
12
<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:sns="http://www.mulesoft.org/schema/mule/sqs"
      xsi:schemaLocation="
               http://www.mulesoft.org/schema/mule/core
               http://www.mulesoft.org/schema/mule/core/current/mule.xsd
               http://www.mulesoft.org/schema/mule/sns
               http://www.mulesoft.org/schema/mule/sqs/current/mule-sqs.xsd">

      <!-- put your global configuration elements and flows here -->

</mule>

Adding the Amazon SQS Connector to a Flow

  1. Create a new Mule project in Anypoint Studio.

  2. Drag the Amazon SQS connector onto the canvas, then select it to open the properties editor.

  3. Configure the connector’s parameters:

    demo_receivemessages

    Field Value

    Display Name

    Enter a unique label for the connector in your application.

    Connector Configuration

    Select a global Amazon SQS connector element from the drop-drown.

    Operation

    Select an operation for the connector perform.

    Queue URL

    Select a parameter for the operation.

  4. Click the blank space on the canvas to save your connector configurations.

Using the Connector in a Mavenized Mule App

If you are coding a Mavenized Mule application, this XML snippet must be included in your pom.xml file.


          
       
1
2
3
4
5
<dependency>
  <groupId>org.mule.modules</groupId>
  <artifactId>mule-module-sqs</artifactId>
  <version>4.0.2</version>
</dependency>

Inside the <version> tags, put the desired version number, the word RELEASE for the latest release, or SNAPSHOT for the latest available version. The available versions to date are:

  • 4.1.0

  • 4.0.2

  • 4.0.1

  • 3.2.0

  • 3.1.1

  • 3.1.0

  • 3.0.0

  • 2.5.5

Demo Flows Using the Connector

Send a message along with meta data to an Amazon SQS queue and then receive it from the queue. This can be split into the following two flows:

  1. Send message along with metadata, and then get the count of the messages in the queue to validate that the message has been sent.

  2. Receive the message, log the message body, and delete the message from the queue.

Studio Visual Editor View


Send Message Operation Flow Receive and Delete Message Operation Flow

Create a Flow to Send a Message

Begin the flow by sending a message to the queue:

  1. Create a new Mule project in Anypoint Studio.

  2. Drag an HTTP Connector into the canvas, then select it to open the properties editor console.

  3. Add a new HTTP Listener Configuration global element:

  4. In General Settings, click the + button:

    amazon_sqs_http_config

  5. Configure the following HTTP parameters, while retaining the default values for the other fields:

    HTTPParams

    Field Value

    Name

    HTTP Listener Configuration

    Port

    8081

  6. Add a Transform Message component to attach the metadata:

    Transform Message Component

    
                 
              
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    
    %dw 1.0
    %output application/java
    ---
    {
            delaySeconds: 0,
            messageBody: "Hello World",
            messageAttributes: {
                    "AccountId": {
                            "stringValue" : "000123456",
                            "dataType" : "String.AccountId"
                    } as :object {
                            class: "org.mule.modules.sqs.model.MessageAttributeValue"
                    },
                    "NumberId": {
                            "stringValue" : "230.000000000000000001",
                            "dataType" : "Number"
                    } as :object {
                            class : "org.mule.modules.sqs.model.MessageAttributeValue"
                    }
            } as :object {
                    class: "java.util.HashMap"
            }
    } as :object {
            class: "org.mule.modules.sqs.model.Message"
    }
  7. Drag an Amazon SQS connector into the flow, and double-click the connector to open its Properties Editor.

  8. If you do not have an existing Amazon SQS connector global element to choose, click the plus sign next to Connector Configuration.

    Demo_ConnectorConfiguration

  9. Configure the global element properties, then click OK.

  10. Configure the remaining parameters of the connector:

    Send Message Parameters

    Field Value

    Display Name

    Enter a name for the connector instance.

    Connector Configuration

    Select the global configuration you create.

    Operation

    Send Message

    Message

    #[payload]

  11. Add an Object To JSON transformer to convert the response from connector into JSON.

  12. Add a Logger to print the response in the Mule Console.

    demo_logger

    Field Value

    Display Name

    Enter a name for the logger.

    Message

    Sent Message: #[payload]

    Level

    INFO (Default)

  13. Add another Amazon SQS connector to get the count of the messages in the queue.

    demo_getmessagecount

    Field Value

    Display Name

    Enter a name for the connector instance.

    Connector Configuration

    Select the global configuration you create.

    Operation

    Get approximate number of messages.

  14. Add a Logger to print the number in the Mule Console.

    demo_logger2

Create a Flow to Receive a Message

This completes the first part of the use case. Now create another flow to receive message and long them before deleting them from the queue.

  1. Drag an Amazon SQS connector and configure it as an inbound endpoint:

    demo_receivemessages

    Field Value

    Display Name

    Enter a name for the connector instance.

    Connector Configuration

    Select the global configuration you create.

    Operation

    Receive Messages

    Number of Messages

    1

    Visibility Timeout

    30

    The Message processor’s Queue URL attribute takes precedence over the Global Element Properties Queue URL. If none of the attributes belonging to Global Element Properties, including Queue Name, Queue URL, and the Message Processor’s Queue URL is provided, the connector throws an exception.

  2. Add a Logger to print the message in the Mule Console:

    Field Value

    Display Name

    Enter a name of your choice.

    Message

    Received Message: #[payload]

    Level

    INFO (Default)

  3. Add another Logger to print the message handle in the console.

    demo_displaymessagehandle

    Field Value

    Display Name

    Enter a name of your choice.

    Message

    Deleting message with handle: #[header:inbound:sqs.message.receipt.handle]

    Level

    INFO (Default)

  4. Now configure an Amazon SQS connector to delete the message from the queue.

    demo_deletemessage

    Field Value

    Display Name

    Enter a name for the connector instance.

    Connector Configuration

    Select the global configuration you create.

    Operation

    Delete Message

  5. Add a Logger to print the status in the mule console after the message is deleted.

Anypoint Studio XML Editor

For this code to work in Anypoint Studio, you must provide Amazon Web Services credentials. You can either replace the variables with their values in the code, or you can provide the values for each variable in the src/main/app/mule-app.properties file.

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:dw="http://www.mulesoft.org/schema/mule/ee/dw" xmlns:json="http://www.mulesoft.org/schema/mule/json" xmlns:http="http://www.mulesoft.org/schema/mule/http"
        xmlns:sqs="http://www.mulesoft.org/schema/mule/sqs" xmlns:tracking="http://www.mulesoft.org/schema/mule/ee/tracking"
        xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
        xmlns:spring="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd http://www.mulesoft.org/schema/mule/sqs http://www.mulesoft.org/schema/mule/sqs/current/mule-sqs.xsd http://www.mulesoft.org/schema/mule/ee/tracking http://www.mulesoft.org/schema/mule/ee/tracking/current/mule-tracking-ee.xsd http://www.mulesoft.org/schema/mule/json http://www.mulesoft.org/schema/mule/json/current/mule-json.xsd http://www.mulesoft.org/schema/mule/ee/dw http://www.mulesoft.org/schema/mule/ee/dw/current/dw.xsd">
        <http:listener-config name="HTTP_Listener_Configuration"
                host="0.0.0.0" port="8081" doc:name="HTTP Listener Configuration" />
    <sqs:config name="Amazon_SQS_Configuration" accessKey="${sqs.accessKey}" secretKey="${sqs.secretKey}" defaultQueueName="${sqs.queueName}" region="${sqs.region}" doc:name="Amazon SQS: Configuration"/>
        <flow name="sqs-send-message-operation-demo-flow">
                <http:listener config-ref="HTTP_Listener_Configuration"
                        path="/sendmessage" doc:name="HTTP" />
        <dw:transform-message doc:name="Transform Message">
            <dw:set-payload><![CDATA[%dw 1.0
%output application/java
---
{
        delaySeconds: 0,
        messageBody: "Hello World",
        messageAttributes: {
                "AccountId": {
                        "stringValue" : "000123456",
                        "dataType" : "String.AccountId"
                } as :object {
                        class: "org.mule.modules.sqs.model.MessageAttributeValue"
                },
                "NumberId": {
                        "stringValue" : "230.000000000000000001",
                        "dataType" : "Number"
                } as :object {
                        class : "org.mule.modules.sqs.model.MessageAttributeValue"
                }
        } as :object {
                class: "java.util.HashMap"
        }
} as :object {
        class: "org.mule.modules.sqs.model.Message"
}]]></dw:set-payload>
        </dw:transform-message>
        <sqs:send-message config-ref="Amazon_SQS_Configuration" doc:name="Send Message">
            <sqs:message ref="#[payload]"/>
        </sqs:send-message>
        <json:object-to-json-transformer doc:name="Object to JSON"/>
        <logger message="Sent Message : #[payload]" level="INFO" doc:name="Display Sent Message"/>
                <sqs:get-approximate-number-of-messages
                        config-ref="Amazon_SQS_Configuration" doc:name="Get Count of Messages in queue" />
        <logger message="Approx. messages in queue : #[payload]" level="INFO" doc:name="Count Messages in Queue"/>
                <set-payload value="Operations successful, check the Mule console for message received ."
                        doc:name="Display Message Count" />
        </flow>
        <flow name="sqs-receive-delete-message-operations-demo-flow">
        <sqs:receive-messages config-ref="Amazon_SQS_Configuration" doc:name="Amazon SQS (Streaming) Receive Messages"/>
                <logger message="Received Message : #[payload]" level="INFO"
                        doc:name="Display Message" />
        <logger message="Deleting message with handle : #[header:inbound:sqs.message.receipt.handle]" level="INFO" doc:name="Display Message Handle"/>
        <sqs:delete-message config-ref="Amazon_SQS_Configuration" doc:name="Delete Message"/>
        <logger message="Message deleted successfully from queue." level="INFO" doc:name="Logger"/>
        </flow>
</mule>

Download Demos

You can download a fully working example from github.io.

Resources