JIRA: https://jira.spring.io/browse/INT-4069
Finding annotations with AnnotatedElementUtils allows us to use
explicit annotation attribute overrides specified by the @AliasFor
annotation.
* Upgrade to the latest Reactor-2.5.0
* Fix all `Promise` and `Streams` mentioning to the `Mono` and `Flux`
* Add `ReactiveEndpoint.SubscribableChannelPublisherAdapter` to adapt `SubscribableChannel` into a `Publisher` for Reactive downstream
* Add `ReactiveMessageChannel` implementation based on the provided `Processor<?>` for internal dispatching logic.
Implement the `Publisher<?>` contract to allow downstream `Subscriber<?>`
* Add `ReactiveEndpoint` implementation to deal with the `ReactiveMessageChannel` through the Reactive Streams contract.
* Add appropriate `ReactiveEndpoint` building logic to the `ConsumerEndpointFactoryBean` and `AbstractMethodAnnotationPostProcessor`
* Rework Gateway to use `Publisher<?>` contract instead of `Promise` implementation.
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**
JIRA: https://jira.spring.io/browse/INT-4067
When `FileSplitter` is configured with `markers = true` and file is empty, an `iterator` for file throws `IOException: Stream closed`,
because we close the `buffer` just after the first `readLine()` attempt, but still return `true` from the first `hasNext()` call
where the `this.sof` and `this.eof` are `true` for markers.
Add logic to mark internal splitter `iterator` as `done` where we don't have content and still in `sof` state.
JIRA: https://jira.spring.io/browse/INT-4013
Previously we couldn't define a config as :
````xml
<int:channel-interceptor pattern="input*">
<ref bean="sampleInterceptor" />
</int:channel-interceptor>
````
* Fix `GlobalChannelInterceptorParser` to distinguish `BeanDefinitionParserDelegate.REF_ELEMENT` child elements
and extract an appropriate `BeanReference`
* Modify `GlobalChannelInterceptorTests-context.xml` to prove the fix and meet JIRA requirements
JIRA: https://jira.spring.io/browse/INT-4057
When general router `channelKey` returns just a `Class<?>` result, we end up with the
`unsupported return type for router [class java.lang.Class]` and forced to to call its `getName()` in the target application code before returning to router.
* Change the `AbstractMappingMessageRouter` logic to treat `Class<?>` as a special String-aware case, use its `getName()` and don't go to the `ConversionService`
* Increase receive timeout for replies in the `TcpInboundGatewayTests`
See https://build.spring.io/browse/INT-AT42SIO-199/
In some places we start `Cache` but doesn't close/destroy it in the end.
That sometimes causes conflicts with the already started GemFire from other tests or wrong state around `BeanFactoryLocator`
* Ensure `Cache` destroy in the end of each test
* In addition destroy `Region` as well to avoid unexpected race conditions, too
* Plus optimize some tests to start `Cache` and `Region` only once per test class, not for each test method
From here I can say that it even would be better to start `Cache`, and therefore whole GemFire, only once per entire Gemfire module test suite.
But that is fully different story.
Remove `forkEvery` from Gemfire module
Allow get a performance improvement for the build.
The `gradlew clean :spring-integration-gemfire:testAll` with existing state is like:
```
Total time: 3 mins 41.375 secs
```
With this fix and without `forkEvery`:
```
Total time: 1 mins 42.699 secs
```
So, 2 min saved time!
Plus fix timeouts in the `PayloadSerializingTransformerParserTests`
JIRA: https://jira.spring.io/browse/INT-1463
An existing `QueueChannel#getRemainingCapacity()` is based on the `BlockingQueue#getRemainingCapacity()` which is always `Integer.MAX_VALUE`
and doesn't reflect reality for provided `capacity`
* Fix `PriorityChannel` to return `upperBound.availablePermits()` for `getRemainingCapacity()`
* Add test for `PriorityChannel` on the matter
* Also increase `reply-timeout` for `<int-redis:queue-outbound-gateway>` in `RedisQueueGatewayIntegrationTests-context.xml` from 1 sec to 10 secs
