---
schema_version: "1"
id: 728d5c35fec9d9b00fd25d30f206f2adb9481a8ebb07efa65883aea275b3f678
title: Migrating to the AMQP Connector
collection: mulesoft
version: current
locale: en-us
portal_url: https://docs.salesforce.com/mulesoft/current/en-us/mule-runtime/4.4/migration-connectors-amqp.md
canonical_url: https://docs.mulesoft.com/mule-runtime/4.4/migration-connectors-amqp
source_site: docs.mulesoft.com
content_type: documentation
content_hash: 00393615398d1c54502e013a40b6bd9450358994489f2518c0f378e64b1559a9
description: MuleSoft Documentation Site
release: "4.4"
latest: false
---
---
product: Mule Runtime
version: 4.4
---

# Migrating to the AMQP Connector

> For the full documentation index, see: https://docs.mulesoft.com/llms.txt

As in other cases the AMQP connector was completely reshaped and evolved away from the Mule 3
transport model into an operation based connector with a simplified UX and fully DataSense enabled
Message information. This means not only that using AMQP now provides the same experience regarding
connection and configuration that any other connector, but also that publishing and consuming
messages is simpler based on the structured Message information.

## Configuring The Connector

Moving from a 3.x transport configuration to a AMQP Connector configuration in Mule 4 implies
basically using most of the same parameters, but declaring them in a more cohesive way. Some
parameters changed in Mule 4 AMQP Connector. This means for example that parameters that are used
for consuming Messages are declared in a 'consumer-config' group, while general authentication
parameters are configured in the General group. Also, a separation has been made between the
parameters that affect the behavior of the connector (present in the config) from those that affect
only how the connection is established (present at connection level):

Mule 3 Example: Configuring The Connector

```xml
<amqp:connector name="AMQP_Config"
     fallbackAddresses="192.168.0.1,192.168.0.2"
     virtualHost="/"
     username="guest"
     password="guest"
	 deliveryMode="TRANSIENT"
	 priority="1"
	 ackMode="AUTO"
	 activeDeclarationsOnly="false"
	 mandatory="false"
	 immediate="false"
	 prefetchSize="16"
	 prefetchCount="16"
	 noLocal="false"
	 exclusiveConsumers="false"
	 requestBrokerConfirms="false"
	 numberOfChannels="10"
	  />
```

This same configuration is achieved in Mule 4 setting the same parameters in the context where they
actually apply:

Mule 4 Example: Configuring The Connector

```xml
<amqp:config name="AMQP_Config">
	<amqp:connection host="localhost" port="5672"
		virtualHost="/" username="guest" password="guest" />
	<amqp:consumer-config numberOfConsumers="10" exclusiveConsumers="false" noLocal="false" ackMode="AUTO"/>
	<amqp:publisher-config requestBrokerConfirms="false" mandatory="false" immediate="false" priority="1" deliveryMode="TRANSIENT"/>
	<amqp:quality-of-service prefetchSize="16" prefetchCount="16" />
</amqp:config>
```

Notice that `fallbackAddresses` is not supported anymore.

## Sending Messages

With the new Mule 4 approach, the AMQP 'publish' operation relies only on its input parameters to
completely build the AMQP Message to be published.

For example, if we wanted to send a high priority AMQP Message with only a part of the payload in
the body, and associate that Message to a group, we’ll need to: <1>) Use `transform` to set the
payload to what the Message body is expected to be. <2>) Convert the resulting stream into a String
to send it as a text message. <3>) Set an AMQP property with `priority` as key to set the AMQP
Message Priority.

Mule 3 Example: Sending a prioritized Message as part of a Group

```xml
<flow name="AmqpTransportOutbound">
    <http:listener config-ref="HTTP_Listener_Configuration" path="/orders"/>
    <dw:transform-message> (1)
        <dw:set-payload><![CDATA[%dw 1.0
%output application/json
---
{
order_id: payload.id,
supplier: payload.warehouse
}]]></dw:set-payload>
    </dw:transform-message>
    <object-to-string-transformer/> (2)
    <amqp:outbound-endpoint exchangeName="testExchange" connector-ref="AMQP_Connector" >
      <message-properties-transformer scope="outbound">
          <add-message-property key="priority" value="9"/> (3)
      </message-properties-transformer>
    </amqp:outbound-endpoint>
</flow>
```

The same results can be achieved in Mule 4 using the AMQP Connector with the following
configuration:

Mule 4 Example: Sending a prioritized Message as part of a Group

```xml
<flow name="AMQPConnectorPublish">
		<http:listener config-ref="HTTP_Listener_config" path="/orders"/>
		(2)
		<amqp:publish config-ref="AMQP_Config" exchangeName="targetExchange">
			<amqp:message> (1)
				<amqp:body>#[output application/json ---
        {
          order_id: payload.id,
          supplier: payload.warehouse
        }]</amqp:body>
        			<amqp:properties priority="3"/> (3)
			</amqp:message>
		</amqp:publish>
	</flow>
```

Differences to be noted:

1. There’s no need of the `transform` component, since the `body` of the Message is created inline,
   thus the payload remains unmodified. 2) The `object-to-string` transformer was also removed,
   since the Connector can automatically handle the transformation output. 3) Priority is set as a
   property in the AmqpMessage operation and does not rely on the user knowing the exact key.

As a summary, when publishing a Message in 3.x with the AMQP transport, we relied on the AmqpMessage
payload, and outbound properties to configure the creation of the AMQP Message, which meant a deeper
knowledge of how the transport worked. In 4.x, the AMQP Connector exposes every configurable element
as a parameter in the scope were it belongs, thus exposing all the AMQP functionality in a clearer
way.

## Listening For New Messages

The AMQP transport `inbound-endpoint` allows you to wait for new AMQP Messages on a given queue. The
output of this listener will contain the body of the message in the payload, and all the AMQP
headers and properties encompassed in the `AMQP Attributes`.

Mule 3 Example: Listening For Messages

```xml
<flow name="AMQPTransportInbound">
  <amqp:inbound-endpoint connector-ref="AMQP_Connector" queueName="in" />
  <dw:transform-message> (2)
      <dw:set-payload><![CDATA[%dw 1.0
        %output application/json
        ---
        {
        items: payload,
        costumer: message.inboundProperties.'costumer_id'
        }]]></dw:set-payload>
  </dw:transform-message>
  <object-to-string-transformer/>  (3)
  <amqp:outbound-endpoint exchangeName="v2/prime/orders" connector-ref="AMQP_Connector"/>  (4)
</flow>
```

In this case, we are listening for Messages and then adapting them to the new format required:

1. Transform the MuleMessage using the metadata contained in the inboundProperties so the payload
   matches the new JSON format we need for the new API. 2) Convert the transformed payload to a JSON
   String. 3) Publish the payload to the defined exchange.

Implementing the same in Mule 4 looks like this:

Mule 4 Example: Listening For Messages

```xml
<flow name="AMQPConnectorPublish">
  <amqp:listener config-ref="AMQP_Config" queueName="in" /> (1)
  <amqp:publish config-ref="AMQP_Config" exchangeName="ordersExchange"> (2)
    <amqp:message>
      <amqp:body>#[output application/json ---
      {
        items: payload,
        costumer: attributes.properties.userProperties.costumer_id, (3)
        type: attributes.headers.type
      }]</amqp:body>
    </amqp:message>
  </amqp:publish>
</flow>
```

Now, the flow has fewer components and is not required to modify the Message payload to publish with
a different format:

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>Definition of the new message is done inline, so it only creates the JSON for the new Message body.</td></tr><tr><td><i data-value="2"></i><b>2</b></td><td>We use the message 'attributes' POJO instead of the 'inboundProperties', which now differentiate the 'headers' of the AMQP Messsage from the 'properties'.</td></tr></tbody></table>

## Consuming Messages

Consuming Messages mid-flow from a given destination was not supported by Mule’s 3 AMQP transport,
and the way to go was also adding the 'Mule Requester Module' to your application, which would then
handle the mid-flow message consume.

So, for example, if you wanted to expose your AMQP Queue, your application would be similar to this:

Mule 3 Example: Consuming Messages Mid-Flow

```xml
<flow name="ordersFromAMQP">
  <http:inbound-endpoint exchange-pattern="request-response" path="orders" host="localhost" port="8081"/>
  <scripting:transformer doc:name="AMQP Message Listening">
    <scripting:script engine="Groovy"><![CDATA[
org.mule.api.MuleMessage message = new org.mule.module.client.MuleClient(muleContext).request('amqp://recordsyntactic_exchange/amqp-queue?connector=AMQP_0_9_Connector&exchangeType=direct&queueDurable=true&exchangeDurable=true&queueAutoDelete=true', 10000L);
]]></scripting:script>
</flow>
```

Some things to notice here are:

- All metadata regarding AMQP Message is completely lost, so logging the CorrelationId relies on you
  knowing the syntax to obtain the header.

- We need both the AMQP and the configuration for the queue in the request.

Mule 4 comes out of the box with the capability of consuming messages mid-flow by using 'consume'
operation. This operation is very similar to the listener we saw before, with the difference that it
can be used anywhere in the flow:

Mule 4 Example: Consuming Messages Mid-Flow

```xml
<flow name="ordersFromAMQP">
  <http:listener config-ref="HTTP_Listener_config" path="/orders"/>
  <amqp:consume config-ref="config"  queueName="Orders" />
  <logger level="INFO" message="#['CorrelationId: ' ++ attributes.properties.correlationId]"/>
</flow>
```

## Doing Request-Reply

AMQP allows you to use the `reply_to` property to perform a implement the RPC pattern. This can be
done either with a temporary temporary exclusive reply queue that is created on the fly by the
client, or using an already existing queue.

### Request Reply With Temporary Auto-delete Private Reply Queue

In Mule 3, for the first case where the reply queue is a temporary exclusive queue that will be
discarded once the message arrives, we have the "request-response" exchange-pattern in the outbound
endpoint:

Mule 3 Example: Doing Request-Reply With Temporary Reply Destination

```xml
<flow name="amqpRequestReplyTemporaryDestination">
  <http:inbound-endpoint exchange-pattern="request-response" host="localhost" port="8080" path="invoices"/>
  <dw:transform-message>
      <dw:set-payload><![CDATA[%dw 1.0
        %output application/xml
        ---
        {
        data: payload,
        costumer: message.inboundProperties."http.query.params".costumer_id
        }]]></dw:set-payload>
  </dw:transform-message>
  <object-to-string-transformer/>
  <amqp:outbound-endpoint exchange-pattern="request-response" queueName="invoiceProcessor" connector-ref="AMQP_Connector"/>
  <logger level="INFO" message="Status: #[payload]">
</flow>
```

Instead, in Mule 4 you have a brand new operation called `publish-consume` which aims to solve this
specific use case:

Mule 4 Example: Doing Request-Reply With Temporary Reply Destination

```xml
<flow name="amqpRequestReplyTemporaryDestination">
  <http:listener config-ref="HTTP_Listener_config" path="/invoices"/>
  <amqp:publish-consume config-ref="AMQP_Config" exchangeName="invoiceProcessor">
    <amqp:message>
      <amqp:body>#[output application/xml ---
      {
        data: payload,
        costumer: attributes.queryParams.costumer_id
      }]</amqp:body>
    </amqp:message>
  </amqp:publish-consume>
  <logger level="INFO" message="#['Status: ' ++ payload]">
</flow>
```

You may see that, again, the building of the Message is done inline of the operation, in the
`message` element, and any transformation or configuration that affects the outgoing Message will be
done as part of that element.

### Request Reply With Explicit Reply To Queue

Doing a request-reply with an explicit `reply-to` queue was can be done in Mule 4 through the
replyTo property:

Mule 4 Example: Doing Request-Reply With Explicit Reply To

```xml
<flow name="amqpRequestReplyTemporaryDestination">
  <http:listener config-ref="HTTP_Listener_config" path="/invoices"/>
  <amqp:publish-consume config-ref="AMQP_Config" exchangeName="targetExchange">
    <amqp:message>
      <amqp:body>#[output application/xml ---
      {
        data: payload,
        costumer: attributes.queryParams.costumer_id
      }]</amqp:body>
      <amqp:properties replyTo="replyToQueue" />
    </amqp:message>
  </amqp:publish-consume>
  <logger level="INFO" message="#['Status: ' ++ payload]">
</flow>
```

## Using Transactions

Transactions support is quite similar in its configuration when moving from 3.x to 4.x, with the
expected change from it being configured in the `inbound-endpoint` and `outbound-endpoint` to the
normalized Mule 4 approach for operations and sources:

Mule 3 Example: Using Transactions

```xml
<flow name="transactedAmqpFlow">
    <amqp:inbound-endpoint queue=Name"${in}">
        <amqp:transaction action="ALWAYS_BEGIN" /> (1)
    </amqp:inbound-endpoint>
    <set-variable variableName="originalPayload" value="#[payload]"/> (2)
    <dw:transform-message> (3)
        <dw:set-payload><![CDATA[%dw 1.0
          %output application/xml
          ---
          payload
          ]]></dw:set-payload>
    </dw:transform-message>
    <object-to-string-transformer/>
    <amqp:outbound-endpoint exchangeName="${out}"> (4)
        <amqp:transaction action="ALWAYS_JOIN"/>
    </amqp:outbound-endpoint>
    <default-exception-strategy>
        <commit-transaction exception-pattern="*"/> (5)
        <set-payload value="#[flowVars.originalPayload]"/> (6)
        <amqp:outbound-endpoint queue="dead.letter"> (7)
            <amqp:transaction action="JOIN_IF_POSSIBLE"/>
        </amqp:outbound-endpoint>
    </default-exception-strategy>
</flow>
```

Things to note are:

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>Transaction is initiated by the inbound endpoint with <code>ALWAYS_BEGIN</code></td></tr><tr><td><i data-value="2"></i><b>2</b></td><td>We make sure not to loose the original payload</td></tr><tr><td><i data-value="3"></i><b>3</b></td><td>Payload is transformed so it can be sent through the outbound endpoint</td></tr><tr><td><i data-value="4"></i><b>4</b></td><td>The outbound endpoint is configured to <code>ALWAYS_JOIN</code></td></tr><tr><td><i data-value="5"></i><b>5</b></td><td>We set up the exception strategy to catch all exceptions</td></tr><tr><td><i data-value="6"></i><b>6</b></td><td>Original payload is restored so the original message is published to the dead.letter</td></tr><tr><td><i data-value="7"></i><b>7</b></td><td>Finally we send the original message to the dead.letter attempting to join to the current transaction.</td></tr></tbody></table>

Same scenario can be implemented in Mule 4 with the following approach:

Mule 4 Example: Using Transactions

```xml
<flow name="transactedAmqpFlow">
    <amqp:listener config-ref="AMQP_Config" queueName="${in}" transactionalAction="ALWAYS_BEGIN"/> (1)
    <amqp:publish config-ref="AMQP_Config" destination="${out}" transactionalAction="ALWAYS_JOIN"> (2)
        <amqp:message>
            <amqp:body>#[output application/xml --- payload</amqp:body>
        </amqp:message>
    </amqp:publish>
    <error-handler>
        <on-error-continue type="ANY"> (3)
          <amqp:publish config-ref="AMQP_Config" exchangeName="dead.letter" transactionalAction="JOIN_IF_POSSIBLE"> (4)
          	<amqp:routing-keys>
				<amqp:routing-key value="dead.letter" />
			</amqp:routing-keys>
          </amqp:publish>
        </on-error-continue>
    </error-handler>
</flow>
```

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>Transaction is initiated by the <code>listener</code> with <code>ALWAYS_BEGIN</code></td></tr><tr><td><i data-value="2"></i><b>2</b></td><td>Publishing of the payload in XML format is done by the <code>publish</code> operation without modifying the current payload, also joining the transaction with <code>ALWAYS_JOIN</code></td></tr><tr><td><i data-value="3"></i><b>3</b></td><td>An error handler that catches any error occurred is used to make sure the Message is not lost</td></tr><tr><td><i data-value="4"></i><b>4</b></td><td>Since the current payload is still the original Message received, we just publish it to the dead.letter using the <code>JOIN_IF_POSSIBLE</code> transactional action</td></tr></tbody></table>

## Enforcing the Existence of Queue and Exchanges

In Mule 3, if you set `activeDeclarationsOnly` to false and the declared queue or exchange does not
exist, the AMQP connector throws a `ShutdownSignalException` involving a 404 error. You can enforce
the existence of a queue or exchange in Mule 4 using the `createFallbackQueue` and
`createFallbackExchange` parameters.

```xml
<amqp:config name="Amqp_Config" createFallbackQueue="false" createFallbackQueue="false">
    <amqp:connection host="localhost" username="guest" password="guest" />
</amqp:config>
```