JIRA: https://jira.spring.io/browse/AMQP-532
Previously, pending confirms were synchronized on the map value (itself a map).
A concurrent modifification exception was reported in `generateNacksForPendingAcks`.
It's not clear how this could happen because all uses synchronize on the map value.
However, since only one thread can use a channel at a time (obtained from the CF), it
is safe to simply synchronize on the channel itself rather than using a complex locking
scheme.
Change synchronization to the channel object; synchronize on the channel in the
`RabbitTemplate.getUnconfirmed()` method.
Add javadocs to `addListener` to tell users they must synchronize on the channel.
In a future release, we should not expose the map - add a note to the Javadoc.
Also, while testing, it was determined that an issue could occur if the channel closes
after the listener is added but before the message was sent.
When the channel is a `PublisherCallbackChannel` throw an exception to the called if
the underlying channel closes; call the channel so it will distribute pending acks as
nacks.
(cherry picked from commit 7770da8)
JIRA:https://jira.spring.io/browse/AMQP-506
`RabbitTemplate` provides a method `getUnconfirmed` to get aged
non-confirmed correlation data.
This was not thread-safe.
While the `pendingConfirms` objects in `RabbitTemplate` and `PublisherCallbackChannelImpl`
are concurrent hash maps, they are maps of maps with the inner maps being
`Collections.synchronizedSortedMap`. While iterating over these maps, they must be
synchronized.
Add a test case to reproduce the exception and verify the problem is resovled.
JIRA: https://jira.spring.io/browse/AMQP-498
In earlier versions, the while loop to drain the queue took an early exit
when the channel was transacted. (This was actually incorrect since message
delivery is not transactional, only acks).
AMQP-388 introduced a new OR condition on the loop: `consumer.hasDelivery()`.
This effectively made the `continuable` boolean in the OR condition irrelevant.
This boolean was true if messages were received and the channel was not transactional.
Another side effect was the `continuable` boolean was incorrectly used to adjust the consumers
if the workload demanded. Instead, that decision should have depended just on whether
messages were received; whether the channel is transacted is irrelevent.
Remove the `continuable` boolean; fix the while loop and use a new boolean `receivedOk` which
indicates whether messages were received and so whether we should consider increasing the
consumers.
Add a test to verify consumers are adjusted for transactional channels.
JIRA: https://jira.spring.io/browse/AMQP-481
Previously the `@Lazy` components haven't been registered with `MessageListenerContainer`,
because `RabbitListenerEndpointRegistrar` has done that from its `afterPropertiesSet()`.
Add logic to `registerListenerContainer` on demand from the `RabbitListenerEndpointRegistrar#registerEndpoint`
**Cherry-pick to 1.4.x**
JIRA: https://jira.spring.io/browse/AMQP-493
Allow the user to generate custom consumer tags instead of the default server-generated tag.
Remove `ServerGeneratesConsumerTagStrategy` and replace it with ternary operator:
`(this.tagStrategy != null ? this.tagStrategy.createConsumerTag(queue) : "")`
JIRA: https://jira.spring.io/browse/AMQP-490
Currently, we suppress the error log if the channel is closed normally.
However, we should also suppress if the channel is closed because the connection
is closed normally.
(cherry picked from commit d21b795)
JIRA: https://jira.spring.io/browse/AMQP-467
Log at INFO level when a new connection is established; useful
when recovering a connection after a broker loss.
JIRA: https://jira.spring.io/browse/AMQP-487
The `ChannelProxy.target` can be closed externally in between several `close` operations on the proxy.
The second `close()` caused an NPE before this fix.
In addition the closed `ChannelProxy` may be returned from the cache. Prevent NPE on the `target.close()` operation.
**Cherry-pick to 1.4.x**
Conflicts:
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
Resolved.
JIRA: https://jira.spring.io/browse/AMQP-484
(must remain at SSLv3 when backporting to 1.4.x)
Also add overridable method for additional context customization.
AMQP-484: Javadoc Polishing
Change `TLSv1.1` to the `SSLv3` by default
JIRA: https://jira.spring.io/browse/AMQP-482
Make the threshold at which `LongString`s are converted as `DataInputStream`s configurable.
Polishing
Revert `whats-new.xml` and `amqp.xml` changes
JIRA: https://jira.spring.io/browse/AMQP-474
Instead of replacing the placeholder on every log event,
replace it once during initialization.
Polishing; Restore ampqProps.appId
AMQP-474: Polishing
Conflicts:
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
JIRA: https://jira.spring.io/browse/AMQP-474
When logs are generated soon after initialization (as is the case for
the test cases), some log messages could (infrequently) be lost.
This is due to the `EventSender` seeing a `null` `applicationId`, generating
a routing key:
[null.org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests.DEBUG]
instead of:
[AmqpAppenderTest.org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests.DEBUG]
(in the case of the logback test case).
Since this does not match a binding, the log message is discarded and the test fails.
Since the `applicationId` setter is called before `start()`, on the same thread, the
only way this can happen is if the `EventSender` runs on a CPU that has a stale
copy of the `AmqpAppender` in its cache (the applicationId field is not marked `volatile`).
Rather than making all the log variables volatile, synchronized blocks are now
used during `EventSender` initialization, to ensure the senders see the variable
values correctly.
JIRA: https://jira.spring.io/browse/AMQP-472
Previously, any `IOException` during passive queue declaration would enter
declaration retry and eventually throw a `QueuesNotAvailableException`. Whether
or not that is recoverable depends on the container's `missingQueuesFatal` property.
If the `IOException` is due to a connection close, we should not try to redeclare
and, further, recovery should be unconditional.
When a queue declaration fails, check if the connection is open and, if not, throw
an `AmqpIOException`, causing container recovery to begin. If the connection is open
continue retrying queue declaration as before.
In addition, expose the queue declaration retry properties on the `SimpleMessageListenerContainer`.
AMQP-472: Polishing
Polishing - PR Comments
AMQP-472: Polishing to the last changes
JIRA: https://jira.spring.io/browse/AMQP-466
Avoid (unlikely, but possible) breaking change.
Deprecate `AddressUtils` altogether.
Revert the impacted code to use only `message.getMessageProperties().getReplyToAddress()`, where the decoding operation is located now.
JIRA: https://jira.spring.io/browse/AMQP-466
1.4.1 introduced direct reply to and reply to address decoding required
using `AddressUtils`. The code should have been put in the `Address` ctor.
JIRA: https://jira.spring.io/browse/AMQP-463
JRockit throws an `IllegalArgumentException` if a
`TreeMap$Entry` is used after removal via the `Iterator`.
The `PublishSubscribeChannelImpl` uses this technique.
Obtain the entry value before removing the entry.
Add a test to reproduce the problem with the JRockit JVM.
Currently sends 10k messages across 100 threads; also tested
with 1m messages across 100 threads with no lost confirms.
JIRA: https://jira.spring.io/browse/AMQP-457
If the listener container is configured to listen to multiple queues,
it would be useful for the listener to have access to which queue a
message was received from.
Instead of a collection of consumer tags, maintain a `Map` of consumer
tags to queue names and populate the message properties with the
tag and queue name.
In the spring-messaging header mapper, map the properties to headers
(inbound only).
JIRA: https://jira.spring.io/browse/AMQP-459
Add configuration such that the `channelCacheSize` can be
considered a limit of how many channels can be created and
a timeout when requesting a channel.
Polishing
JIRA: https://jira.spring.io/browse/XD-2079
Use the `ContainerDelegate`'s class loader to create the proxy.
Problems in XD creating the proxy to apply the advice chain
in the rabbit source.
JIRA: https://jira.spring.io/browse/AMQP-456
Previously all abnormal channel closes were logged at ERROR level,
even NOT FOUND for passive declarations.
JIRA: https://jira.spring.io/browse/AMQP-453
Add optional compression (gzip, zip) to the `RabbitTemplate` and
`SimpleMessageListenerContainer`.
- MessagePostProcessors
-- support gzip, zip compression, decompression
- RabbitTemplate
-- new properties `beforePublishPostProcessor`, `afterReceivePostProcessor`
-- configure compression on send, decompression on receive
- Listener Container
-- support decompression
- New Header
-- `springAutoDecompress`, required for backards compatibility (so we don't decompress existing messages)
- Test cases
Fix Javadocs
AMQP-453 Polishing - PR Comments
- Support customization of the deflating streams
- Support ordered lists of post processors
Revert to Simple Subclasses
- Add support for compression level
- Default compression level is BEST_SPEED
AMQP-453 Add Tests to Verify Level
Refactoring Names
JIRA: https://jira.spring.io/browse/AMQP-451
- Add `flush()` to support flushing any partial batch.
- Implement `Lifecycle` and invoke `flush()` when stopped.
JIRA: https://jira.spring.io/browse/AMQP-446
Previously, 3 WARN log entries were generated when declaring
a non-durable, auto-delete, exclusive queue. 2 logs were generated
for a non-durable, auto-delete exchange.
Reduce log level to INFO and only emit 1 log per queue/exchange.
JIRA: https://jira.spring.io/browse/AMQP-451
Released in stealth mode for now (no docs, javadocs marked `Experimental`).
This feature is to support XD; we can't incubate in XD because it needs a
change to the listener container.
AMQP-451 Add Perf Test
Move Debatching Code
Move from `SMLC.doReceiveAndExecute()` to `AMLC.executeListener()`.
- Change the cause of the `ListenerExecutionFailedException` on a bad
decode to a `MessageConversionException` so that the default error handler
will cause the message to be rejected - otherwise the badly formed message
will continually be redelivered by default.
- Add a test case to ensure a badly formed message is rejected.
JIRA: https://jira.spring.io/browse/AMQP-448
RabbitMQ now supports "Direct reply-to" where an RPC sender
can specify `amq.rabbitmq.reply-to` and the broker provides
special handling to return the reply to the sending channel.
This is more efficient than creating a temporary reply queue each
time. The `RabbitTemplate` has provided an alternative to creating
new reply queues via a reply message listener container and correlation
data.
With this change, the template now supports "Direct reply-to" and will
use it instead of temporary reply queues when no reply-listener is
provided.
Tested on a server with 3.4.1 RabbitMQ as well as one with 3.1.1 for
backwards compatibility.
__NOTE:__ On the receiving side, the message `reply_to` property
has the form `amq.rabbitmq.reply-to.<base64string>`. Because `/` is a
valid `base64` character, we can't use normal reply-to address decoding
`<exchange>/<routingKey>`. The `MessageListenerAdapter` therefore detects
that Direct reply-to is being used before decoding the address.
Polishing
Move reply address decoding to a static method in `Address` and invoke from all places
where it is needed.
Causes a minor leak of rabbitmq into spring-amqp (the name of the direct reply queue).
AMQP-448 Polishing; PR Comments
JIRA: https://jira.spring.io/browse/AMQP-438
Since `DefaultExceptionStrategy` treats only `org.springframework.amqp.support.converter.MessageConversionException` as `fatal`
and assuming backward compatibility for SF < 4.1, add `catch (org.springframework.messaging.converter.MessageConversionException ex) {`
to the `MessagingMessageListenerAdapter` to wrap that exception to the `org.springframework.amqp.support.converter.MessageConversionException`.
Having that the `DefaultExceptionStrategy` works with `@RabbitListener` as it is with generic `<rabbit:listener>`.
JIRA: https://jira.spring.io/browse/AMQP-435
Add ...NoWait methods to the PublisherCallbackChannelImpl.
Invoke the wait versions for now; add TODOs to change to the
...NoWait versions when 3.4.0 is the minimum client.
AMQP-435 Polishing - Add getDelegate()
Throw UnsupportedOperationException telling users to invoke
via getDelegate().
Add `What's New` notice
JIRA: https://jira.spring.io/browse/AMQP-436
Use the list of queue names as the lookup key.
Polishing `testAbstractRoutingConnectionFactoryWithListenerContainer()` to reflect the dynamic usage
MessageProperties
JIRA: https://jira.spring.io/browse/AMQP-432
Previously `DefaultMessagePropertiesConverter` converted just
`LongString`,
but BasicProperties may contain List and Map values with nested
LongStrings. Values unsupported by AMQP which are contained in Lists or
Maps are also not converted from `MessageProperties` to
`BasicProperties`.
Add logic to properly convert from LongString values nested in
Lists/Maps.
Add logic to properly convert unsupported values nested in Lists/Maps to
Strings.
**Cherry-pick to 1.3.x**
Code style polishing
JIRA: https://jira.spring.io/browse/AMQP-406
* Add support of `connection-factory-selector-expression` for the `RabbitTemplate`
and provide logic to bind target `ConnectionFactory` to the invocation context based on the result of the expression evaluation.
AMQP-406: Fix double ThreadLocal resource
Polishing
AMQP-406: introduce `receiveConnectionFactorySelectorExpression`
Doc Polishing
