AMQP-414: Rabbit annotation-driven endpoints

JIRA: https://jira.spring.io/browse/AMQP-414

This commit adds the support of Rabbit annotated endpoint. Can be
activated by both @EnableRabbit or <rabbit:annotation-driven/>
and detect methods of managed beans annotated with @RabbitListener,
either directly or through a meta-annotation.

Containers are created and managed under the cover by a registry
at application startup time. Container creation is delegated to a
RabbitListenerContainerFactory that is identified by the
containerFactory attribute of the RabbitListener annotation. Containers
can be retrieved from the registry using a custom id that can be
specified directly on the annotation.

The configuration can be fine-tuned by implementing the
RabbitListenerConfigurer interface which gives access to the registrar
used to register endpoints. This includes a programmatic registration
of endpoints in complement to the declarative approach. A default
RabbitListenerContainerFactory can also be specified to be used if no
containerFactory has been set on the annotation.

Annotated methods can have flexible method arguments that are similar
to what @MessageMapping provides. In particular, rabbit listener
endpoint methods can fully use the messaging abstraction, including
convenient header accessor. It is also possible to inject the raw
org.springframework.amqp.core.Message and the Channel for more
advanced use cases. The payload can be injected as long as the
conversion service is able to convert it from the original body. By
default, a DefaultMessageHandlerMethodFactory is used but it can be
configured further to support additional method arguments or to
customize conversion and validation support.

The return type of an annotated method can also be an instance of
Spring's Message abstraction. Instead of just converting the payload,
such response type allows to communicate standard and custom headers.

AmqpHeaders and DefaultAmqpHeaderMapper have been copied from
Spring integration. SimpleAmqpHeaderMapper is a reduced version of
the Spring integration counter part that does not support deprecated
fields and JSON specific constructs. Method hook points are available
so that a future version of Spring Integration can integrate those
customizations in an extension. Note that the default know maps all
user-defined headers so that the generated Message abstraction has
all the information stored in the protocol specific message.

MessagingMessageConverter is a new MessageConverter implementation
that produces a org.springframework.messaging.Message out of an AMQP
message and vice versa. AmqpMessageHeaderAccessor provides an easy
access of AMQP specific headers from a standard Message.

Issue: AMQP-414

AMQP-414: Polishing
AMQP-421: Fix Memory Leak with Publisher Confirms

JIRA: https://jira.spring.io/browse/AMQP-421

The wrong key is being used for `RabbitTemplate.pendingConfirms`. The rabbit template
stores the pending confirms map under the `ChannelProxy` whereas the
removal uses the actual `PublisherCallbackChannel` so the map
entry is not removed.

Use the correct key to store the pending callback map so that it
is removed from the template when closed.

Change a test case to use the `CachingConnectionFactory` instead of
a mock and verify the map is cleared.

**Add docs explaining that Publisher Confirms and Returns only works
with cached channels - ensure the cache size is sufficiently large.
AMQP-417 CoExist With AutoRecoveryEnabled

JIRA: https://jira.spring.io/browse/AMQP-417

When the underlying connection factory has `autoRecoveryEnabled` set to
`true`, the client maintains a list of open channels (in
AutorecoveringConnection). When the client is able to reconnect, it
re-establishes these channels.

Channels are removed from the list of recovery candidates when they
are `close()`d.

These mechanisms are incompatible with the existing `CachingConnectionFactory`
and the `SimpleMessageListenerContainer`.

First, when the caching connection factory detects its connection is closed,
it attempts to create a new connection. It is not aware of the old 'recovered'
connection, which ends up leaked.

Second, when physically closing a channel (always done by container consumers),
`close()` is only called if `isOpen()` returns true - hence the "closed"
channel is not removed from the recovery list, but it is no longer in the
cache so the channel is leaked. Worse, the associated blocking queue will
receive messages after recovery, but since the consumer is no longer active
those messages will remain in the queue unacked.

Similarly, cached channels, when retrieved from the cache, are silently
dropped from the cache if `isOpen()` returns false. Again, these remain
as recovery candidates, but are no longer available in the cache.

USERS ARE ADVISED TO NOT USE AUTORECOVERYENABLED WITH SPRING AMQP

Full resolution will require changing the architecture of the listener
container.

This commit provides a work around for those users that do set the option;
in essence, it disables channel recovery for those channels that are managed
by Spring.

- `SimpleConnection.isOpen()` always returns true for `AutorecoveringConnection`s
     This prevents the caching connection factory from creating a new connection
     - it will reuse this connection when recovered.
- When pysically closing channels, don't check `isOpen()`. Instead, always call
     `close()` and catch any `AlreadyClosedException`. This forces the
     channel to remove itself (unregister) from the `AutoRecoveringConnection`'s
     recoverable channels.

This is difficult to create an automated integration test. A test with mocks
veriifies that the channel is physically closed (run the test without the code
changes and it fails - `close()` is not called).

Also tested with the Spring Integration AMQP sample application. With
`autoRecoveryEnabled`, restarting the server causes 2 connections and 3
channels. With the commit, the recovery is disabled and a single connection
(auto-recovered) is established, with the consumer and producer channels
(the producer channel is only created when you send a message).
AMQP-417 Polishing, Fix Broken Tests

When the factory is destroyed, close all channels.

A number of tests using mocks failed now that the channels are
unconditionally closed.

Change the tests to use a CachingConnectionFactory so the channels
are cached instead of physically closed.