[[putting-it-all-together]]
= Putting it All Together

The following configuration creates an exchange `myDestination` with queue `myDestination.consumerGroup` bound to a topic exchange with a wildcard routing key `#`:

[source]
---
spring.cloud.stream.bindings.input.destination=myDestination
spring.cloud.stream.bindings.input.group=consumerGroup
#disable binder retries
spring.cloud.stream.bindings.input.consumer.max-attempts=1
#dlx/dlq setup
spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true
spring.cloud.stream.rabbit.bindings.input.consumer.dlq-ttl=5000
spring.cloud.stream.rabbit.bindings.input.consumer.dlq-dead-letter-exchange=
---

This configuration creates a DLQ bound to a direct exchange (`DLX`) with a routing key of `myDestination.consumerGroup`.
When messages are rejected, they are routed to the DLQ.
After 5 seconds, the message expires and is routed to the original queue by using the queue name as the routing key, as shown in the following example:

.Spring Boot application
[source,java]
----
@SpringBootApplication
public class XDeathApplication {

    public static void main(String[] args) {
        SpringApplication.run(XDeathApplication.class, args);
    }

    @Bean
    public Consumer<Message<String>> listen() {
        return message -> {
            Map<?,?> death = message.getHeaders().get("x-death");
            if (death != null && death.get("count").equals(3L)) {
                // giving up - don't send to DLX
                throw new ImmediateAcknowledgeAmqpException("Failed after 4 attempts");
            }
            throw new AmqpRejectAndDontRequeueException("failed");
        };
    }

}
----

Notice that the count property in the `x-death` header is a `Long`.

[[rabbit-error-channels]]
= Error Channels

Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel.
See "`<<spring-cloud-stream-overview-error-handling>>`" for more information.

RabbitMQ has two types of send failures:

* Returned messages,
* Negatively acknowledged https://www.rabbitmq.com/confirms.html[Publisher Confirms].

The latter is rare.
According to the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.".
You can also get a negative acknowledgment if you publish to a bounded queue with `reject-publish` queue overflow behavior.

As well as enabling producer error channels (as described in "`<<spring-cloud-stream-overview-error-handling>>`"), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows:

* `ccf.setPublisherConfirms(true);`
* `ccf.setPublisherReturns(true);`

When using Spring Boot configuration for the connection factory, set the following properties:

* `spring.rabbitmq.publisher-confirms`
* `spring.rabbitmq.publisher-returns`

The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessageException` with the following properties:

* `failedMessage`: The spring-messaging `Message<?>` that failed to be sent.
* `amqpMessage`: The raw spring-amqp `Message`.
* `replyCode`: An integer value indicating the reason for the failure (for example, 312 - No route).
* `replyText`: A text value indicating the reason for the failure (for example, `NO_ROUTE`).
* `exchange`: The exchange to which the message was published.
* `routingKey`: The routing key used when the message was published.

Also see <<publisher-confirms>> for an alternative mechanism to receive returned messages.

For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageException` with the following properties:

* `failedMessage`: The spring-messaging `Message<?>` that failed to be sent.
* `nackReason`: A reason (if available -- you may need to examine the broker logs for more information).

There is no automatic handling of these exceptions (such as sending to a <<rabbit-dlq-processing, dead-letter queue>>).
You can consume these exceptions with your own Spring Integration flow.

[[rabbit-binder-health-indicator]]
= Rabbit Binder Health Indicator

The health indicator for Rabbit binder delegates to the one provided from Spring Boot.
For more information on this, see https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#actuator.endpoints.health.auto-configured-health-indicators[this].

You can disable this health indicator at the binder level by using the property - `management.health.binders.enabled` and set this to `false`.
In the case of multi-binder environments, this has to be set on the binder's environment properties.

When the health indicator is disabled, you should see something like the below in the health actuator endpoint:

```
"rabbit": {
  "status": "UNKNOWN"
}
```

At the Spring Boot level, if you want to disable the Rabbit health indicator, you need to use the property `management.health.rabbit.enabled` and set to `false`.
