Free MuleSoft CONNECT Keynote & Expo Pass Available!

Register now+
Nav

Amazon SQS Connector

Select

The Anypoint Connector for Amazon SQS provides an easy way to interface with the Amazon Simple Queue Service API, allowing Mule users to manage SQS queueing services without having to deal with the API directly.

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.

Amazon SQS connector provides support for two types of queues "Standard Queue" and "FIFO Queue". Standard queues have high throughput but occasionally more than one copy of a message can be delivered and messages can be delivered in an order different from which they are sent. FIFO queues have limited throughput, but messages are delivered in exactly the same order in which they are sent.

Prerequisites

This document assumes that you are familiar with Mule, Anypoint Connectors, Anypoint Studio essentials, elements in a Mule flow, and global elements.

As the connector comes prebundled 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 5.0 is compatible with:

Application/Service Version

Mule Runtime

4.0.x or later

AWS SDK for Java

1.11.79

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 expects a Map<String, MessageAttributeValue> object, which requires you to refer to the org.mule.extension.sqs.api.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.

Creating FIFO Queue

To create the FIFO queue you have to use the 'create queue' operation in the connector and add two attributes additionaly. Those two attibutes are FifoQueue as true and ContentBasedDeduplication as true. The queue name should end with .fifo suffix. Example is 'MyTestFIFOQueue.fifo'.

While creating the FIFO queues you have to choose either of the two regions US East (Ohio) or US West (Oregon) as Amazon only supports these two regions for FIFO queues.

While using the SQS Connector to send messages to FIFO queue, you should mention the same queue URL and region name that you used while creating the FIFO Queue. Also you have to give the value to the attribute 'message group id' in the connector configuration while sending messages.

To Install this Connector

  1. In Anypoint Studio, click the Exchange icon in the Studio taskbar.

  2. Click Login in Anypoint Exchange.

  3. Search for the connector and click Install.

  4. Follow the prompts to install the connector.

When Studio has an update, a message displays in the lower right corner, which you can click to install the update.

Updating Your POM File

When a new version of a connector is released, you have to add the POM dependency in your mule application POM.xml file. For exaxmple:

<dependency>
    <groupId>org.mule.connectors</groupId>
    <artifactId>mule-sqs-connector</artifactId>
    <version>5.0.0</version>
    <classifier>mule-plugin</classifier>
</dependency>

To Create a New Project

  1. In Studio, click File > New > Mule Project.

  2. Enter a name for your new project and leave the remaining options with their default values.

    amazon-sqs-new-project

  3. If you plan to use Git, select Create a .gitignore file for the project with default ignores for Studio Projects, and then click Next.

  4. Click Finish to create the project.

Configuring the Amazon SQS Connector Global Element

To use the Amazon SQS connector in your Mule application, configure a global element that can be used by all the Amazon SQS connectors in the application.

To create a global Amazon SQS connector configuration:

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

  2. On the Global Configuration Elements screen, click Create.

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

    amazon-sqs-global-type

  4. Click OK.

  5. Enter the global element properties:

    amazon-sqs-config-pic

    Field Description

    Access Key

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

    Secret Key

    Key that plays the role of a password.

    Try AWS Credentials Provider Chain

    Dropdown 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.

+ . Keep the Proxy and the Advanced tabs with their default entries. . 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. . 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. 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

  • Read (Receive Messages)

  • Remove permission

  • Send message batch

  • Send message

  • Set Queue Attributes

Adding the Amazon SQS Connector to a Studio Flow

  1. Create a new Mule project in Anypoint Studio.

  2. Select the Amazon SQS connector and drag the operation Read onto the canvas, then select it to open the properties editor.

  3. Configure the operation parameters:

    amazon-sqs-demo-receive-messages

    Field Value

    Display Name

    Enter a unique label for the connector operation in your application.

    Connector Configuration

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

    Queue URL

    Select a parameter for the operation.

    Max no of messages

    Give value to a parameter for the operation.

  4. Save your connector configuration.

Example Use Case

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.

Studio Visual Editor

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 Listener operation into the canvas, then select it to open the properties editor console.

  3. Add a new HTTP Listener Configuration global element.

  4. In the General group for the item, set Path to the / value.

  5. In Basic Settings, click the plus button:

    amazon-sqs-http-config

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

    amazon-sqs-http-params

    Field Value

    Host

    0.0.0.0

    Port

    8081

  7. 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 2.0
    output application/java
    ---
    {
            delaySeconds: 0,
            body: "Hello World",
            messageAttributes: {
                    "AccountId": {
                            "stringValue" : "000123456",
                            "dataType" : "String.AccountId"
                    } as Object {
                            class: "org.mule.extension.sqs.api.model.MessageAttributeValue"
                    },
                    "NumberId": {
                            "stringValue" : "230.000000000000000001",
                            "dataType" : "Number"
                    } as Object {
                            class : "org.mule.extension.sqs.api.model.MessageAttributeValue"
                    }
            } as Object {
                    class: "java.util.HashMap"
            }
    } as Object {
            class: "org.mule.extension.sqs.api.model.Message"
    }
  8. Drag an Amazon SQS connector send message operation into the flow, and double-click the connector to open its Properties Editor.

  9. In Basic Settings on the right side of the field Extension configuration click the plus icon and add values for Access Key, Secret Key, and Queue Name.

  10. Configure the remaining parameters of the connector.

    Send Message Parameters

    Field Value

    Display Name

    Enter a name for the connector operation.

    Connector Configuration

    Select the global configuration you create.

    Message

    #[payload]

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

    amazon-sqs-demo-logger

    Field Value

    Display Name

    Enter a name for the logger.

    Message

    Sent Message: #[payload]

    Level

    INFO (Default)

  12. Add another Amazon SQS connector > Get approximate number of messages to get the count of the messages in the queue.

    amazon-sqs-demo-get-message-count

    Field Value

    Display Name

    Enter a name for the connector operation.

    Connector Configuration

    Select the global configuration you create.

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

    amazon-sqs-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 log them before deleting them from the queue.

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

    amazon-sqs-demo-receive-messages

    Field Value

    Display Name

    Enter a name for the connector operation.

    Connector Configuration

    Select the global configuration you create.

    Number of Messages

    10

    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

    #[payload]

    Level

    INFO (Default)

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/resources/mule-artifact.properties file.


          
       
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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core" xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns:sqs="http://www.mulesoft.org/schema/mule/sqs" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 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/sqs http://www.mulesoft.org/schema/mule/sqs/current/mule-sqs.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/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd">
  <configuration-properties file="mule-artifact.properties"/>
  <sqs:config name="Amazon_SQS_Configuration" doc:name="Amazon SQS Configuration" doc:id="ID_VALUE">
    <sqs:basic-connection accessKey="${sqs.accessKey}" secretKey="${sqs.secretKey}" defaultQueueName="${sqs.queueName}"/>
  </sqs:config>
  <http:listener-config name="HTTP_Listener_config" doc:name="HTTP Listener config" doc:id="ID_VALUE">
    <http:listener-connection host="0.0.0.0" port="8081"/>
  </http:listener-config>
  <flow name="sqs-send-message-operation-demo-flow" doc:id="ID_VALUE">
    <http:listener config-ref="HTTP_Listener_config" path="/" doc:name="Listener" doc:id="ID_VALUE"/>
    <ee:transform doc:name="Transform Message" doc:id="ID_VALUE">
      <ee:message>
        <ee:set-payload><![CDATA[%dw 2.0
output application/java
---
{
        delaySeconds: 0,
        body: "Hello World",
        messageAttributes: {
                "AccountId": {
                        "stringValue" : "000123456",
                        "dataType" : "String.AccountId"
                } as Object {
                        class: "org.mule.extension.sqs.api.model.MessageAttributeValue"
                },
                "NumberId": {
                        "stringValue" : "230.000000000000000001",
                        "dataType" : "Number"
                } as Object {
                        class : "org.mule.extension.sqs.api.model.MessageAttributeValue"
                }
        } as Object {
                class: "java.util.HashMap"
        }
} as Object {
        class: "org.mule.extension.sqs.api.model.Message"
}]]></ee:set-payload>
      </ee:message>
    </ee:transform>
    <sqs:send-message config-ref="Amazon_SQS_Configuration" doc:name="Send message" doc:id="ID_VALUE" message="#[payload]"/>
    <logger level="INFO" doc:name="Logger" doc:id="ID_VALUE" message="#[payload]"/>
    <sqs:get-approximate-number-of-messages config-ref="Amazon_SQS_Configuration" doc:name="Get approximate number of messages" doc:id="ID_VALUE"/>
    <logger level="INFO" doc:name="Logger" doc:id="ID_VALUE" message="#[payload]"/>
  </flow>
  <flow name="sqs-receive-delete-message-operations-demo-flow" doc:id="ID_VALUE">
    <sqs:receivemessages config-ref="Amazon_SQS_Configuration" doc:name="Receivemessages" doc:id="ID_VALUE"/>
    <logger level="INFO" doc:name="Logger" doc:id="ID_VALUE message="#[payload]"/>
  </flow>
</mule>