JIRA: https://jira.spring.io/browse/INT-4126
Looks like `WATCH` doesn't provide good isolation when the call is performed concurrently at the same time from different Threads.
* Change locking algorithm as it is recommended by Redis `SET` command: http://redis.io/commands/set:
```
SET resource-name anystring NX EX max-lock-time
```
* Add `@Repeat(10)` to the `AggregatorWithRedisLocksTests#testDistributedAggregator()` since that helped to reproduce the issue
**Cherry-pick to 4.3.x & 4.2.x**
Conflicts:
spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
Resolved.
JIRA: https://jira.spring.io/browse/INT-4113
* To simplify `MessagePublishingErrorHandler` and avoid extra `PollerMetadata` beans, added the `errorChannel()` attribute to the `@Poller` annotation
* The `MessagePublishingErrorHandler` now supports the late-binding via new `defaultErrorChannelName`
* Docs about new `errorChannel()` attribute
* Some other docs polishing
Polishing
**Cherry-pick to 4.3.x**
Polishing docs according PR comments
Conflicts:
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
Resolved and reworked Java8 features in tests.
JIRA: https://jira.spring.io/browse/INT-4124
`ByteBuffer.array()` returns null when Direct.
Change `ChannelInputStream` to use the `ByteBuffer` directly instead of
the underlying `byte[]`.
Conflicts:
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java
* Change Lambda for `EventPublisher` to `Mock`
JIRA: https://jira.spring.io/browse/INT-4120
INT-3899 moved creating the pollingTask to start() to allow modification
of the advice chain between starts.
Since a new pollingTask is created on each start, this was fine for general advices.
However, for `AbstractMessageSourceAdvice`s, which only advise the `receive()` operation,
the advices are re-applied on every start.
When starting, first remove any advices we added on the previous start.
This handles the case when the source is a naked object, or already a proxy when supplied
to the SPCA.
Remove uneeded cast
Polishing - PR Comments
JIRA: https://jira.spring.io/browse/INT-4105
Fixes GH-1888
The lock in Redis can be expired in between `obtain()` calls.
So, even if we return a new lock instance, the old one must clear properly.
* Add `lock.unlock()` to the `obtain()` if the case of expiration in Redis.
* Add `warn` for the exception on the `lock.unlock()`. We don't care about error here and just proceed to a new instance.
**Chery-pick to 4.3.x**
Remove the lock reference from `weakThreadLocks` as well
Add `registry.setUseWeakReferences(true);` to test-case
* Add assert for `lock.unlock()` in case of expiration
* Add `UUID.randomUUID()` for the registry key to avoid cross-talking during concurrent builds, e.g. on the CI server
https://build.spring.io/browse/INT-MASTER-352
This is actually a fix for the
JIRA: https://jira.spring.io/browse/INT-4083
JIRA: https://jira.spring.io/browse/INT-4096
* Introduce `spring.integration.readOnly.headers` Integration property
* Introduce `IntegrationMessageHeaderAccessor.setReadOnlyHeaders()` and use it from the overridden `IntegrationMessageHeaderAccessor.isReadOnly()`
* Add `DefaultMessageBuilderFactory.setReadOnlyHeaders()` and delegate the value to the new `MessageBuilder.readOnlyHeaders()`
* Modify `spring.integration.properties` for the `contentType` as a `readOnly` header for testing
* Ensure that provided logic works via an appropriate modification to the `ObjectToJsonTransformerParserTests`
Increase receive timeouts in the `OutboundGatewayFunctionTests`
Fix `FileInboundTransactionTests` for slow `WatchService` issue on Linux/OS X
Also redo the `tmp` dir logic to the `TemporaryFolder` `@ClassRule`
Fix unused `import` in the `FileInboundTransactionTests`
Polishing
JIRA: https://jira.spring.io/browse/INT-4108
Some `Lifecycle.start()/stop()` usage doesn't ensure robustness for components causing unexpected and difficulty tracing issues
* Fix `Lifecycle.start()/stop()` for `FileReadingMessageSource`, `FileWritingMessageHandler`, `AbstractMqttMessageHandler`
* In the `DefaultHeaderChannelRegistry`, `LockRegistryLeaderInitiator`, `MqttPahoMessageHandler` rework logic for shared variables to avoid `NPE`
* Increase receive timeouts in the `PayloadDeserializingTransformerParserTests` and `UdpChannelAdapterTests`
* Prove with the `WatchServiceDirectoryScannerTests` changes that several invocation for `FileReadingMessageSource.start()` are idempotent
**Cherry-pick to 4.3.x**
JIRA: https://jira.spring.io/browse/INT-4102
There is inconsistency when we mark `JmsMessageDrivenEndpoint` for `autoStartup = false`,
but external container can still be started by the application context, because of it own default `autoStartup = true`.
* Mark `container` `autoStartup = false` in the `JmsMessageDrivenEndpoint` ctor to align lifecycles.
**Cherry-pick to 4.3.x**
https://build.spring.io/browse/INT-SI43X-6/
Since Exponential Moving Average algorithm is fully based on the time in between events, it is very difficult to reach the stable tests.
All of them used to fail from time to time on the slow CI server.
Mark all Exponential Moving Average Tests with `@Ignore`.
The manual testing is required in case of some changes.
**Cherry-pick to 4.3.x**
(cherry picked from commit 76bb3ad)
https://build.spring.io/browse/INT-AT42SIO-253/
```
java.lang.AssertionError: Should have been able to lock with zookeeper server restarted!
at org.junit.Assert.fail(Assert.java:88)
at org.junit.Assert.assertTrue(Assert.java:41)
at org.springframework.integration.zookeeper.lock.ZkLockRegistryTests.voidLockFailsWhenServerDown(ZkLockRegistryTests.java:339)
```
When we do `testingServer.restart();` we should give more time to let it started on background, therefore increase the next `lock2.tryLock()` to 10 seconds.
**Cherry-pick to 4.3.x and 4.2.x**
(cherry picked from commit 6ffdd5e)
JIRA: https://jira.spring.io/browse/INT-4082
Starting with Spring Data 1.9 the `MappingContextIsNewStrategyFactory` relies on a newly introduced `PersistentEntities` and doesn't register entities lazily any more.
Such a change finishes with the `Unsupported entity` exception when an auditing is switched on (`<mongo:auditing/>`)
for the `MongoDbMessageStore` and `AbstractConfigurableMongoDbMessageStore` internally created `MongoTemplate` and `MessageWrapper` and `MessageDocument` SI internal entities.
* Don't register `ApplicationContext` into internally created `MongoTemplate`s since to avoid entity events emitting for `MessageWrapper` and `MessageDocument`
* Pull `MongoDbMessageBytesConverter` to the top-level class to let customize `MappingMongoConverter` properly if there is need to audit `MessageDocument` anyway,
what can be possible via external injections into `AbstractConfigurableMongoDbMessageStore` implementation
* Fix `mongodb.adoc` for some typos
**Cherry-pick to 4.2.x**
JIRA: https://jira.spring.io/browse/INT-4087
Fixes GH-1863 (https://github.com/spring-projects/spring-integration/issues/1863)
If ZK Client loses the connection to ensemble, it tries to reconnect on the next operation and does that exactly for the `connectionTimeoutMs` and provided `RetryPolicy`.
According to the `Lock.tryLock(long time, TimeUnit unit)` contract we can wait for the lock only during provided `timeout`.
* Introduce `mutexTaskExecutor` and perform `ZK.forPath()` command in the separate `Thread` and wait for the `Future` during provided timeout.
* Perform `mutex.acquire()` only after successful `Future` result and only for the remained timeout.
**Cherry-pick to 4.2.x**
Add `setMutexTaskExecutor()`
Fix Checkstyle typo
Javadoc
* Fix several typos in log messages. And some test on the matter as well
* Add comment to `AbstractPersistentAcceptOnceFileListFilter.rollback()` to clarify the reason of `rollingBack` variable
* Make `RemoteFileTemplate.StreamHolder` as `static` to avoid extra internal variable to outer class instance
* Replace `MessagingException` with `AbstractInboundFileSynchronizingMessageSource` in `init()` method of some components. It isn't Messaging yet in that phase
* Fix `SubscribableRedisChannel.MessageListenerDelegate` to handle `Object` not `String`, because with the `serializer` injection there is no guaranty that incoming is always `String`
* Move `JSch.setLogger(new JschLogger());` in the `DefaultSftpSessionFactory` to `static` block. It really should be done only once
* Remove `Assert.isTrue(this.port >= 0)` from the `DefaultSftpSessionFactory`. The subsequant `initJschSession()` convert it to default `22` port
* Change in the `JschProxyFactoryBean` `UnsupportedOperationException` to `IllegalArgumentException`. Wrong enum is wrong argument. That isn't a problem of operation
* Simplify `stop()` in the `CuratorFrameworkFactoryBean` and mark it as a `this.running = false`. Otherwise it wasn't able to be restarted
* Expose `leaderEventPublisher` in the `LeaderInitiatorFactoryBean` and fix `stop(Runnable callback)` with propagation `callback` to delegate.
Fix `SubscribableRedisChannelTests` for new `handleMessage(Object)` signature
Two initiators for the same path, same `SmartLifecycleRoleController` and, finally, same `adapter`.
So, one initiator after `yield()` stops the `adapter` and at the same time another starts it.
Since there is no barrier in between events and assertion, we end up with an early "re-granting".
* Add `CountDownLatch yieldBarrier` to `countDown()` after performing second `adapter.isRunning()` assert
* `LeaderEventPublisher` waits for the `yieldBarrier` after the first `OnRevokedEvent`
JIRA: https://jira.spring.io/browse/INT-4078
Even if it doesn't hurt to `copyHeadersIfAbsent()`, when it just adds a new headers and doesn't override existing like sequence details,
it doesn't sound reasonable for `Resequencer` to modify the message before and after its resequence logic.
* Change `shouldCopyRequestHeaders()` to `false` for `ResequencingMessageHandler`
JIRA: https://jira.spring.io/browse/INT-4064
We configure `IdempotentReceiver` via `<int:idempotent-receiver>` component or `@IdempotentReceiver` annotation.
In case of regular Java config, e.g. direct `ConsumerEndpointFactoryBean` usage or Java DSL,
it isn't possible to configure `idempotentReceiverInterceptor` enough easy
* Introduce `if...else` logic into `ConsumerEndpointFactoryBean` to proxy `MessageHandler`,
if `adviceChain` contains an newly-introduced `HandleMessageAdvice`.
And do that independently if `MessageHandler` is `AbstractReplyProducingMessageHandler`
* Make `idempotentReceiverInterceptor extends HandleMessageAdvice`
* Skip `HandleMessageAdvice` in the `AbstractReplyProducingMessageHandler`
* Add advice applying logic into the `AbstractMethodAnnotationPostProcessor` as well
Introduce `HandleMessageAdvice` marker interceptor to cover the case when an `Advice` can be advices as well.
Remove unused `setAdviceChainIfPresent()` method in the `AbstractMethodAnnotationPostProcessor`
Document `HandleMessageAdvice`
Increase wait latch timeouts in the `LockRegistryLeaderInitiatorTests`
Doc Polishing
JIRA: https://jira.spring.io/browse/INT-4063
Since `DefaultXmlPayloadConverter` is used from many out-of-the-box components by default,
make it more flexible with the supported input types which can be converted into `Document`
Improve `xml.adoc` about this types and also fix a lot of typos there as well
Address PR comments
JIRA: https://jira.spring.io/browse/INT-4070
The `MessagingGatewaySupport.receive()` is fully based on the `this.messagingTemplate.receiveAndConvert(replyChannel, null);`
which really just extracts `payload` from the `Message` via default `MessageConverter`.
Therefore `@Gateway` code, which expects to poll exactly `Message<?>` from the flow, is invalid at runtime with `ClassCastException`
* Expose `MessagingGatewaySupport.messagingTemplate` property as `protected` to give access for inheritors and siblings
* In the `GatewayProxyFactoryBean` use `messagingTemplate` and `replyChannel` directly from the `MethodInvocationGateway`
to invoke raw `gateway.messagingTemplate.receive(replyChannel)` bypassing any conversion and to avoid breaking changes.
* Proof the solution with `GatewayProxyFactoryBean.testReceiveMessage()`
**Consider to cherry-pick (backport) down to 3.0.x**
Add `messagingGateway.convertReceiveMessage` global property to let revert to previous behavior
* Make `messagingGateway.convertReceiveMessage=true` by default
* Mock `BeanFactory` in the `GatewayProxyFactoryBeanTests.testReceiveMessage()`
to be sure that `messagingGateway.convertReceiveMessage=false` works as expected
* Add `testReceiveMessageConvert()` to demonstrate `ClassCastException`
Rebase and make `convertReceiveMessage` integration property as `false` by default
Introduce `MessagingGatewaySupport.receiveMessage()` and use it from the `GatewayProxyFactoryBean`
JIRA: https://jira.spring.io/browse/INT-4072
When `publishSubscribeChannel` is with `applySequence = true`, a `messageToSend` is overridden with `sequenceDetails` using `MessageBuilder`, therefore a new fresh `Message`.
In case of state propagation, e.g. `SecurityContextPropagationChannelInterceptor`, we just lost the state from the `ThreadStatePropagationChannelInterceptor.MessageWithThreadState` because of new `Message<?>`
* Add into `BroadcastingDispatcher` the logic to delegate `pushSequenceDetails` into `MessageWithThreadState` directly do not lose the `state`
* Make `ThreadStatePropagationChannelInterceptor` as `MessageBuilderFactory`-aware and use it to rebuild an `original` `Message<?>` in the `MessageWithThreadState` to populate `SequenceDetails`
**Cherry-pick to 4.2.x**
Provide an explicit order for `publishSubscribeChannel` subscribers
Fixes GH-1847 (https://github.com/spring-projects/spring-integration/issues/1847)
Fix mutation in the `ThreadStatePropagationChannelInterceptor`
Since `BroadcastingDispatcher` invokes `pushSequenceDetails` for each subscribed handler,
make `MessageWithThreadState` as immutable and return a new instance via `cloneWithSequenceDetails()` method with particular `sequenceDetails`.
Previous mutable solution ended up with the issue of concurrent modification.
* Introduce `CloneableMessage` abstraction to let any custom `Message` to return `MessageBuilder` with desired context.
* Introduce `DelegatingMessageBuilder` as an extension of the `MessageBuilder` to let custom `CloneableMessage` to return desired customization.
* Add into `MessageBuilder#fromMessage()` `if` for the `CloneableMessage`
* Add into `MutableMessageBuilder` a `warn` about `CloneableMessage`
* Revert changes in the `BroadcastingDispatcher` in favor of `CloneableMessage` in the `MessageBuilder`
* Redo `ThreadStatePropagationChannelInterceptor#MessageWithThreadState` logic to be based on the `CloneableMessage` and `DelegatingMessageBuilder` extension.
Introduce `MessageDecorator` contract
Remove `CloneableMessage` aspect and everything around
`MessageWithThreadState` is now `MessageDecorator` and `BroadcastingDispatcher` check if incoming `message` is `MessageDecorator` and performs its `decorateMessage` after `builder`
JIRA: https://jira.spring.io/browse/INT-4075
Fixes GH-1854 (https://github.com/spring-projects/spring-integration/issues/1854)
The `catch` block just swallows an Exception with the logger and leaves `stompSessionListenableFuture` as `null`.
The next `stompSessionListenableFuture` usage over two code lines bellow ends up with `NPE`.
* Add reconnect logic into that `catch` block and just return from there, since we don't have anything to go ahead.
* Add `if (e != null) {` around `log.error` in the `scheduleReconnect`
* Add `if...else` around `this.stompClient.getTaskScheduler()` and `log.info()` to notify end-user that there won't be reconnection if there is no `taskScheduler`
* Add `StompSessionManagerTests` mock tests to verify solution
**Cherry-pick to 4.2.x**
JIRA: https://jira.spring.io/browse/INT-4074
If the `@ServiceActivator` component is processed first the application will start.
However, if another component that relies on the channel creation goes first it'll fail as the bean for the channel does not exist.
* Add `name` setters for all channels in the `GatewayProxyFactoryBean` to allow late channel resolution in the target `MethodInvocationGateway`
* Change `MessagingGatewayRegistrar` to populate channel properties as names not bean references
**Cherry-pick to 4.2.x**
