There is no `com.rabbitmq:http-client` (Hop) dependency anymore.
The `WebClient` is used internally in the project for tests.
Mention `WebClient` sample in the doc instead.
* Fix `broker-configuration.adoc` for the `RabbitAdmin.getQueueInfo()` API
* Move existing `What's New` to the respective `changes-in-3-2-since-3-1.adoc`
* Add entries for the current version into `What's New`
* Start new `rabbitmq-amqp-client.adoc` chapter
Fixes: #3006
We had an issue in our system trying to consume a message from RabbitMQ that contained an `x-death` header, with count value typed as `Integer`.
This caused a `ClassCastException`.
The reason the count value was an int and not a long is that we were storing headers in an internal database as part of a recovery process, and the typing was slightly changed during serialisation / deserialisation.
* Use `target.setRetryCount(numberValue.longValue());` in the `DefaultMessagePropertiesConverter` instead of cast to `long`
Signed-off-by: Raul Avila <raul.avila@pagonxt.com>
[artem.bilan@broadcom.com Improve commit message]
**Auto-cherry-pick to `3.2.x`**
Signed-off-by: Artem Bilan <artem.bilan@broadcom.com>
The logic if the `processQ1AndQ2Data()` listener method is to
collect 10 messages, where one of them is a duplication after `requeue`.
The first `discard` is accepted into `received` result, but the second is
terminated by the exception.
However, `countDown()` in the `finally` block is still called for all the cases.
Therefore, `11` times: 7 for normal data, 1 for `discard`, 2 for `requeue`, and 1 for exception.
Fixes: #3005
Issue link: https://github.com/spring-projects/spring-amqp/issues/3005
The `SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.killOrRestart()`
is also called during application context shutdown.
At this moment we cannot emit events into an application context.
Otherwise, it fails with:
```
Exception in thread "rabbitListenerExecutor1" org.springframework.beans.factory.BeanCreationNotAllowedException: Error creating bean with name 'refreshEventListener': Singleton bean creation not allowed while singletons of this factory are in destruction (Do not request a bean from a BeanFactory in a destroy method implementation!)
```
* Introduce `ObservableListenerContainer.isApplicationContextClosed()`
and call it as additional condition in the `SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.killOrRestart()`
before trying to emit `AsyncConsumerStoppedEvent`
**Auto-cherry-pick to `3.2.x`**
The fix for `3.1.x` requires a slightly different approach via `ContextClosedEvent`
The `messageProperties.getReplyTo()` might be null, so set it into `amqpMessage::to`
only if it is not null
* Add convenient `JavaUtils.acceptOrElseIfNotNull()` for two alternative values
The `RabbitAmqpUtils.toAmqpMessage()` utility is used on the publisher side,
so, it is natural to treat such a message as a reply.
Therefore, the `correlationId` is set to `messageId` of `correlationId` is not present.
The `replyTo` of the Spring message is set into `to` of the AMQP message
* Mention in the `Address` JavaDocs that just `routingKey` can be treated differently by clients
* Fix error message in the `RabbitAmqpMessageListenerAdapter`
According to the `Collections.synchronizedList()` Javadocs,
it has to be iterated with a `synchronized (list)`.
Otherwise, there is no guarantee that memory barrier for list items is fulfilled.
Fixes: https://github.com/spring-projects/spring-amqp/issues/3002
* Implement `sendAndReceive` & `receiveAndReply` operations in the `RabbitAmqpTemplate`
* Expose contracts to the `AsyncAmqpTemplate`
* Some NullAway fixes for `AsyncAmqpTemplate` hierarchy
* Move DLQ objects for testing to the common `RabbitAmqpTestBase`
Fixes: https://github.com/spring-projects/spring-amqp/issues/2991
It turns out to be not very convenient when application fails on initialization due to not having connection with the broker.
So, the `AmqpConnectionFactoryBean` is wrong contract and better look into a `ConnectionFactory` abstraction instead.
* Rework `AmqpConnectionFactoryBean` with its eager real connection to the `AmqpConnectionFactory` contract and
`SingleAmqpConnectionFactory` implementation where real connection only happens when its `getConnection()` is called
* Rework all the client components to rely on the new `AmqpConnectionFactory` contract
* Make `Publisher` instantiation in a similar to `Connection` manner - on demand, when `RabbitAmqpTemplate` calls its internal `getPublisher()`
Fixes: https://github.com/spring-projects/spring-amqp/issues/3001
* Expose batch related options for `RabbitAmqpListenerContainer` and `RabbitAmqpListenerContainerFactory`, respectively
* `batchSize` - the indicator that `RabbitAmqpListenerContainer` (and `RabbitAmqpMessageListenerAdapter`) has to work in batch mode
* `batchReceiveTimeout` - how long to wait for batch to be fulfilled or release whatever was gathered so far, even if just only one message
* `taskScheduler` - schedule "force batch release" after `batchReceiveTimeout`
* Make `MessagingMessageListenerAdapter` `final` properties as `protected` to avoid undesired copy-paste burden
Fixes: https://github.com/spring-projects/spring-amqp/issues/2994
* Add `@RabbitListener` tests for `RabbitAmqpMessageListener`
* Adjust `RabbitAmqpMessageListener` error handling logic to call `errorHandler` first
* Move `consumer.pause()` for `RabbitAmqpMessageListener.stop()` to the `supplyAsync()`
to initiate pause for all consumers in parallel
* Expose a failed `message` to the `ListenerExecutionFailedException` from the `RabbitAmqpMessageListenerAdapter`
* Ensure RabbitMQ objects are deleted in the end of tests
Related to: https://github.com/spring-projects/spring-amqp/issues/2994
* Add `RabbitAmqpMessageListener` for RabbitMQ AMQP 1.0 native message consumption
* Add `RabbitAmqpMessageListenerAdapter` for `@RabbitLister` API
* Add `RabbitAmqpListenerContainer` and respective `RabbitAmqpListenerContainerFactory`
* Add `AmqpAcknowledgment` as a general abstraction.
In the `RabbitAmqpListenerContainer` delegates to the `Consumer.Context` for manual settlement
* Extract `RabbitAmqpUtils` for conversion to/from AMQP 1.0 native message
* Add convenient `ContainerUtils.isImmediateAcknowledge()` and `ContainerUtils.isAmqpReject()` utilities
* Expose `AmqpAcknowledgment` as a `MessageProperties.amqpAcknowledgment` for generic `MessageListener` use-cases
* Remove `io.micrometer` dependecies from the `spring-rabbitmq-client` module since metrics and observation handled
thoroughly in the `com.rabbitmq.client:amqp-client`
Not tests for the listener yet.
Therefore no fixing for the issue.
Fixes: #2941
Issue link: https://github.com/spring-projects/spring-amqp/issues/2941
Now `BlockingQueueConsumer.basicCancel()` performs `RabbitUtils.closeMessageConsumer()`
to initiate `basicRecovery` on the transactional consumer to re-queue all the un-acked messages.
However, there is a race condition when one in-flight message may still be delivered
to the listener and then TX commit is initiated.
There a `basicAck()` is initiated.
However, such a tag might already be discarded because of the previous `basicRecovery`.
Therefore, adjust `BlockingQueueConsumer.commitIfNecessary()` to skip `basicAck()` if locally transacted and already cancelled.
Right, this may lead to the duplication delivery, but having abnormal shutdown situation we cannot guarantee that this message
to commit has been processed properly.
Also, adjust `BlockingQueueConsumer.nextMessage()` to rollback a message if consumer is canceled instead of
going through the loop via listener
* Increase `replyTimeout` in the `EnableRabbitReturnTypesTests` for resource-sensitive builds
