JIRA: https://jira.spring.io/browse/INT-4193
When using the internal `SequenceAwareMessageGroup` within an correlating
message handler, the messages were copied to a new collection before
checking for duplicate sequences in `canAdd()`.
This was unnecessary since we never add anything to this group, if `canAdd()`
returns true, the message is added to the store; this group is discarded.
Instead, use the message collection from the original group; although it is
not modifiable, this is not an issue because we don't need to modify it.
Conflicts:
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java
src/reference/asciidoc/aggregator.adoc
* Polishing `aggregator.adoc` to reflect reality
https://build.spring.io/browse/INT-MASTER-478
The `EmitterProcessor` proves to be unstable in between subscribers which come and go away from time to time.
Looks like buffered events are dropped when we don't have subscriber just after the previous one has been unsubscribed
* Change the logic in the `IntegrationFlowDefinition.toReactivePublisher()` to call `ReactiveConsumer.adaptToPublisher(MessageChannel)` directly to return provided `Publisher<?>` to the caller for his own responsibility.
This way we don't have any buffering or prefetching in the Framework because we don't do any `Subscription.request(n)` explicitly
* Prove the fix with polishing to the `ReactiveStreamsTests.testPollableReactiveFlow()`
Right now we don't poll `QueueChannel` if there is no demand, therefore we don't need extra `CountDownLatch` to wait for second subscriber.
Plus the logic now remains as a `queue` manner: consumers are concurrent for the data in the `QueueChannel`
Extra `@Repeat(10)` for confirmation.
Previously it failed sporadically, what is confirmed with that one more CI failure
JIRA: https://jira.spring.io/browse/INT-4181
* To avoid duplicate code move actual `Metrics` resolution to the `IntegrationManagementConfigurer` and delegate from the `IntegrationMBeanExporter` for backward compatibility
* Minor refactoring and improvements in the `IntegrationMBeanExporter`
* Fix typos in the `jms/AsyncGatewayTests`
**Cherry-pick to 4.3.x**
https://build.spring.io/browse/INT-MASTER-473/
There is a race condition when we perform `onSubscribeCall()`, but the actual subscription isn't happened yet, therefore a new data for the `EmitterProcessor` is dropped because there is no subscriber yet to drain the buffer
JIRA: https://jira.spring.io/browse/INT-4182
To avoid extra allocation for the `SimpleDateFormat` use a new Java 8 `DateTimeFormatter` based on new `Temporal` abstraction
Leave test-cases as is to save some time for other tasks
Fix RedisLockRegistry for the proper formatter and ZoneId
https://build.spring.io/browse/INT-MASTER-468/
Looks like demand in the `FluxTake` isn't honored:
```
public void request(long n) {
if (wip != 0) {
s.request(n);
}
else if (WIP.compareAndSet(this, 0, 1)) {
if (n >= this.n) {
s.request(Long.MAX_VALUE);
}
else {
s.request(n);
}
}
}
````
It isn't clear why we request from upstream `Long.MAX_VALUE`, if we have only strong `n` limit
So, revert to the `secondSubscriberLatch` and don't send the next message until the subscription from the second `Flux`
https://build.spring.io/browse/INT-MASTER-467/
Even if `EmitterProcessor` looks good for us not requesting until real `subscriber()`, it doesn't keep backlog for late subscribers like it happens some times in our subscriber in the separate Thread.
The solution should be considered as tentative because the real one must be based on the fact that `AMPH` has to perform `subscription.request(n)` as a fact of the subscription from downstream.
In other words mid-flow components should work as passive subscribers and be based on downstream demand
Change all the `BUILD-SNAPSHOT`s to their latest Milestones
Fix compatibility with those Milestones
Upgrade to Spring AMQP 2.0.0.M1
Upgrade to Spring Data Kay
Gemfire now is based on the Apache Geode, so changed all the imports to proper new packages
MongoDB now is based on the Mongo 3 Driver, therefore many breaking changes. Mostly to the mapping part
Fix Checkstyle violation
Fix race condition in the `MongoDbInboundChannelAdapterIntegrationTests` around `QueueChannel` and tx commit
Add `-s` to Travis Gradle command to see stack trace about `MongoDbMetadataStoreTests` problem
Test only MongoDB module on Travis with -d
Add addon to Travis to pull MongoDB-3.0
Fix `MongoDbAvailableRule` for MongoDB 3.0 Driver style
Increase `serverSelectionTimeout` to `100` in the `MongoDbAvailableRule`.
Looks like `0` isn't good value to get immediate answer
The `BridgeHandler` is Reactive `Subscriber` with `MAX_VALUE` demand just `onSubscribe()`.
That causes a drain of the upstream `QueueChannel` to the `ReactiveChannel` with `DirectProcessor`.
The last one just drops messages if there is no subscribers
`EmitterProcessor` doesn't request upstream until real subscriber arrives to it.
Therefore change `BridgeHandler` logic in the `IntegrationFlowDefinition.toReactivePublisher()` to the `EmitterProcessor` to allow late `Subscriber`s and don't lose messages from the `QueueChannel` if there is no downstream `Subscriber`s
JIRA: https://jira.spring.io/browse/INT-4174
I considered just using the channel counter but I suppose there is a (small) possibility
that someone might possibly use a custom dispatcher and subscribe to it directly.
Polishing -- ** master only **
Eliminate conditional logic in AbstractMessageChannel by adding getHandlerCount() to
MessageDispatcher.
Do not cherry pick.
* Also fix JPA tests model for proper database `sequence` to avoid race condition when the test data is inserted with lower `id` than expected predefined data has
https://build.spring.io/browse/INT-MASTER-444/
Looks like we still need `method.setAccessible(true)` in the `LambdaMessageProcessor` even if method is `public`.
I think the problem is because any Lambda is still **inline class**, so even if method is `public`, the class isn't
Fix CheckStyle violations and one more `isLambda()`` in the `RecipientListRouterSpec`
Add `Assert` for method ambiguity to the `LambdaMessageProcessor` to be sure that we get deal only with the Function Interface impl
Since the DSL code is new in this version there is no reason to keep deprecated method
* Remove deprecated `resequence()` and `aggregate()` in the `IntegrationFlowDefinition`
* Remove `InternalAggregatingMessageHandler` with the `MessageGroupProcessorWrapper` delegation logic in favor of newly introduced `AbstractCorrelatingMessageHandler.setOutputProcessor()`
* Rework `Promise` Gateway Docs to the `Mono`
* Rename `AsyncGatewayTests` "promise" words to "mono"
* Resolve `com.rabbitmq.client.FlowListener` deprecation
RecipientListRouter DSL refactoring
Previously there was a DSL specific `DslRecipientListRouter` to overcome the lack of setters in the target `RecipientListRouter` and its `Recipient`
* Add `channelName` variant for the `Recipient`
* Add several new `addRecipient()` methods to the `RecipientListRouter`
All those improvements allow to avoid extra bridge (or adapter) component between Java DSL and target `RecipientListRouter`
* Make `LambdaMessageProcessor` and `GatewayMessageHandler` as `public` classes and move them to the appropriate packages
Address PR comments
Add `isLambda()` condition to `RecipientListRouterSpec` to avoid ambiguity for the provided `GenericSelector` impl
checkstyle polishing
JIRA: https://jira.spring.io/browse/SPR-14948
The `StandardTypeConverter` no longer has a `defaultConversionService` field, it uses
a static instance from `DefaultConversionService`.`
https://sonar.spring.io/component_issues?id=org.springframework.integration%3Aspring-integration%3Amaster#resolved=false|types=BUG
In `IntegrationFlowRegistration` double check locking is ok for `inputChannel`
because we're assigning an existing object, but `MessagingTemplate` constructs
a new object for which double check locking doesn't work.
In any case, for both these items, the chance of concurrent access is extremely low
and is idempotent anyway, so remove double check locking.
Several inner classes can be static.
Other minor fixes.
JIRA: https://jira.spring.io/browse/INT-4158
* Add JavaDocs
* De-Boot `JpaDslTests`
* Fix typos and generics inconsistency in the `JpaOutboundGatewayFactoryBean`
Address PR comments
* Get rid of `JpaOutboundGatewayFactoryBean` usage in the `JpaBaseOutboundEndpointSpec`
* Rework `JpaBaseOutboundEndpointSpec` and its inheritors logic to use `JpaOutboundGateway` directly
* Rename to the `JpaTests`
* Fix JavaDoc in the `IntegrationFlowDefinition`
* Do not use `jpaParameters` in the `JpaExecutor` if it is empty collection, not only null
JIRA:https://jira.spring.io/browse/INT-4138
Expressions for topic, qos, retained.
Also change inbound mapping to `RECEIVED_...` headers.
Fix some minor asciidoc problems in (s)ftp.
Rework Qos/Retained Expressions/Defaults
Encapsulate the logic entirely in the converter.
Polishing - PR Comments
* Add `ConsumerSubscriber` to adapt `Consumer<?>` to the `Subscriber<?>`
* Add `SubscribableChannelPublisherAdapter` to adapt `SubscribableChannel` to `Publisher<?>` via `Flux.create()` on subscription
* Add `PollableChannelPublisherAdapter` to adapt `PollableChannel` to `Publisher<?>` via `Mono.delayMillis()` and `flatMap()` for `channel.receive()`
Use `concatMap()` for `PollableChannel`
Iterate `PollableChannel` until there is a data
Fix Lambda signature error via explicit class declaration:
```
java.lang.ClassFormatError: Duplicate field name&signature in class file org/springframework/integration/endpoint/ReactiveConsumer$PollableChannelPublisherAdapter$1
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(ClassLoader.java:760)
at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at org.springframework.integration.endpoint.ReactiveConsumer$PollableChannelPublisherAdapter.lambda$new$2(ReactiveConsumer.java:163)
```
Looks like Java bug:
Move `Iterator<Message<?>>` instance to the `subscribe()` to avoid race conditions when we have several subscribers.
In other words make `Iterator<Message<?>>` subscriber-specific, but at the same time avoid re-instantiation for each `Flux.concatMap()` call caused by the previous `repeat()`
Upgrade to Reactor 3.0.3 and others
* Make fixes according Reactor 3.0.3 changes
* Remove redundant `TestSubscriber` in favor of `StepVerifier` and `mock(Subscriber)`
* Rework `PublisherIntegrationFlow` Reactive Streams implementation to the out-of-the-box `ReactiveConsumer` and `ReactiveChannel`
JIRA: https://jira.spring.io/browse/INT-4159
Port over DSL for the Mail module
- make tests compatible with the s-i-test TestMailServer
- add `simpleContent` boolean to inbound spec
Polishing - add tx() to ImapIdleChannelAdapterSpec
More Polishing
Rename `Mail.fromMail()` to `Mail.toStringTransformer()` to better reflect reality
JIRA: https://jira.spring.io/browse/INT-4154
* Move `Transformers` for files to the `Files` factory
* Remove `@Deprecated` methods
* Fix `FilesTests` for new packages and removed methods like `.handleWithAdapter()`
Provide JavaDocs for File DSL and implement some new methods like `chmod()` and `renameFunction()`
Fix race condition in the `ScriptsTests` when even small time window delay may lead us to one more message in the queue
JIRA: https://jira.spring.io/browse/INT-4153
* Remove deprecated `FeedFetcher` usage
* Introduce `Resource` based ctor for the `FeedEntryMessageSource`
* Add `SyndFeedInput` option and short-hand `preserveWireFeed` for internal `SyndFeedInput` instance
* Reflect the changes in the XSD for Feed
* Change ROME dependency from deprecated `rome-fetcher` to just `rome` as it is recommended by ROME team
* Port Java DSL for Feed module and reflect aforementioned changes in the `Feed` factory and `FeedEntryMessageSourceSpec` as well
* Document changes and mention Feed Java DSL, too
JIRA: https://jira.spring.io/browse/INT-4012
Previously to configure a `priority` for the `MessageChannel` we should configure `QueueChannel` for particular `MessageStore`.
Although the target priority logic is really in the `MessageStore` implementation, it isn't so obvious why we can't use `PriorityChannel` instance for `MessageStore` case as well.
* Add `PriorityCapableChannelMessageStore` and `MessageGroupQueue` based ctors to the `PriorityChannel` for consistency.
* Delegate the logic to the super `QueueChannel` as before
* Rework `PointToPointChannelParser` and Java DSL components to reflect a new state of the `PriorityChannel`
* Improve Docs on the matter
JIRA: https://jira.spring.io/browse/INT-4163
When `MessageSource` is proxy, the `TransactionSynchronizationManager.getResource(this)` logic in the `MessageSource` doesn't work,
because TX resource is bound to the `Proxy` in the `SourcePollingChannelAdapter`
* Introduce `SourcePollingChannelAdapter.originalSource` property and store there a target `MessageSource` object extracted from the AOP Proxy
* Use `originalSource` as a resource to bind to the TX
* Modify `MongoDbInboundChannelAdapterIntegrationTests` to ensure that `AbstractMessageSourceAdvice` proxying the `MessageSource` doesn't effect `TransactionSynchronizationManager.getResource(this)` logic
* Refactor for some MongoDb test to rely on the `@RunWith(SpringJUnit4ClassRunner.class)` for context loading for better test class performance
**Cherry-pick to 4.3.x, 4.2.x**
Fallback to provided source if `target` from Proxy is `null`
Fix [UnusedImport] issue
https://build.spring.io/browse/INT-MASTER-415/
The `Stream` `Collectors.toList()` returns empty list if nothing pass the `.filter()`, therefore condition as `if (receiveOnlyAdviceChain != null)` is not enough
and since `SourcePollingChannelAdapter.applyReceiveOnlyAdviceChain()` doesn't have conditions as well, the target `MessageSource` is proxyed for nothing.
When `TransactionSynchronizationManager.getResource(this)` is called for the `MessageSource` it can't find it because the proxy doesn't match an original object.
* Make condition as `if (!CollectionUtils.isEmpty(receiveOnlyAdviceChain))` in the `AbstractPollingEndpoint` and `SourcePollingChannelAdapter`
* Increase group removal wait timeout in the `gemfire.DelayerHandlerRescheduleIntegrationTests`
JIRA: https://jira.spring.io/browse/INT-3770,
https://jira.spring.io/browse/INT-4107
Having `TransactionHandleMessageAdvice` we can start TX from any `MessageHandler.handleMessage()`
* Add `<transactional>` alongside with the `<request-handler-advice-chain>` for those components which produce reply
* Merge `<transactional>` and `<request-handler-advice-chain>` configuration to a single `ManagedList`
* Rework JPA `<transactional>` in favor of common solution
* Some polishing and refactoring
AbstractPollingEndpoint: avoid `new ArrayList` if we don't have `receiveOnlyAdvice`s
JIRA: https://jira.spring.io/browse/INT-4157
- Refactor the core message-driven endpoint to be `MessageProducerSupport`, eliminating the DSL wrapper
- Retain and refactor the DSL gateway wrapper - still required because it needs to be a MGS
- Port the (core) `Channels` factory
Fix Javadoc
Remove Deprecated setter
Move JmsInboundGateway to the jms Package
* Some polishing for `@Copyright`s
* Remove `@Deprecated` methods in the `Jms`
* Reinstate `IntegrationFlows.from(Function<Channels, MessageChannelSpec<?, ?>>)`
* Typos fixes in the `JmsTests`
JIRA: https://jira.spring.io/browse/INT-4133
Mostly copy/paste and changes according Java 8 and Reactor 3.0 foundations
Fix `IntegrationFlow` JavaDocs
Increase timeout in the `AbstractCorrelatingMessageHandlerTests`
Polishing - reduce timeout after first expire attempt
JIRA: https://jira.spring.io/browse/INT-4091
When `empty-group-min-timeout` is configured and `expireGroupsUponCompletion == false` and normal or partial sequences group release happens,
schedule the group for removal after `empty-group-min-timeout`, since it is empty already.
That lets to avoid `MessageGroupStoreReaper` configuration just for cleaning empty groups.
* Retrieve the group `lastModified` to check if group is still valid for removal
* Remove `ScheduledFeature` in the remove task
* Polishing for debug messages to reflect the current logic
* Reschedule empty group removal task in case of `InterruptedException` on the `lock.lockInterruptibly()`
* Fix typos in docs
Fix race condition between `groupStore.expireMessageGroups()`and `TaskScheduler` in the `AbstractCorrelatingMessageHandlerTests.testReaperReapsAnEmptyGroupAfterConfiguredDelay()`
Also check groupSize for removal decision
Address PR comments
JIRA: https://jira.spring.io/browse/INT-4151
If the `MessageSource` is already a proxy, we only advise the `receive()` method.
If it's not, we advise all methods, which is incorrect.
* Use `NameMatchMethodPointcutAdvisor` in all advising cases
* Prove with the test case that only `receive()` method is advised for the `MessageSource` proxy
Polishing - PR Comments and Closeable Warnings
Eclipse emits bogus warnings with exceptions in lambdas.
Even though the lambda might run on another thread, elipse thinks it could
cause the context to not be closed.
SPR-14854: MessageChannel is now a @FunctionalInterface
* Additional Lambda polishing and some code style fixes
JIRA: https://jira.spring.io/browse/INT-3502
For some use-cases it is really useful to filter `@MessagingGateway` components as any other `@ComponentScan` capable.
* Add `useDefaultFilters()`, `includeFilters()` and `excludeFilters()` to the `@IntegrationComponentScan`
* Modify `GatewayInterfaceTests` to reflect changes and be sure that logic is applied properly
JIRA: https://jira.spring.io/browse/INT-4144
Fixes GH-1936 (https://github.com/spring-projects/spring-integration/issues/1936)
Previously `LoggingHandler` could be used as a standalone object, without any Spring Container initialization.
Even if that doesn't sound reasonable, we should reinstate the logic to avoid breaking changes
* Initialize `expression` and `evaluationContext` during object `<init>` phase
* Some code reformatting to avoid duplicate blocks and cyclomatic complexity
* Simple code formatting
* Additional JavaDocs for `LoggingHandler`
**Cherry-pick to 4.3.x**
JIRA: https://jira.spring.io/browse/INT-2460,
https://jira.spring.io/browse/INT-4122
Since the main purpose of the `MessageStore` to persist message for durability and only,
it doesn't make sense to modify `Message` for additional headers like `SAVED` and `CREATED_DATE`.
Such a logic should be a part of metadata stored together with the message.
And it is provided by the out-of-the-box `MessageStore` implementation.
In addition we free ourselves from the reflection operations to retain `ID` and `TIMESTAMP` headers when we add `SAVED` and `CREATED_DATE`
* Control "already saved" logic in the `JdbcMessageStore`s via `DuplicateKeyException` on the `INSERT`.
This is much effective then additional `SELECT` in case of `SAVED` before
* Control "already saved" logic in the `AbstractConfigurableMongoDbMessageStore` via `DuplicateKeyException` on the `INSERT`.
Since `MongoDbMessageStore` doesn't provide extra `messageId` field, perform extra `SELECT` before store document.
Anyway the `MongoDbMessageStore` isn't recommended for use.
We may consider to deprecate it
* Control "already saved" logic in the `AbstractKeyValueMessageStore` via `putIfAbsent` operation
With this fix we persist message in the store as is without any modifications when we perform standard serialization procedure.
Any custom serializers should consider to use `MutableMessageBuilder` if there is a requirement to retain `ID` and `TIMESTAMP`
Rework `MongoDbMetadataStore.putIfAbsent()` to normal `findAndModify()` with particular `$setOnInsert`.
Technically the MongoDB query looks like:
```
db.collection.findAndModify({
query: { _id: $key },
update: {
$setOnInsert: { value: $value } // perform modification only on upsert
},
new: false, // don't return new doc if one is upserted
upsert: true // insert the document if it does not exist
})
```
Move single import to the appropriate JavaDoc
Polishing after rebase
DEBUG messages in `doStoreIfAbsent()` implementations
* To keep track of the extra message information in the `MessageStore`, without `Message` modification, introduce `MessageMetadata` and `MessageHolder`
* Add `MessageStore#getMessageMetadata()`
* Modify MongoDb `MessageStore` to add extra `timestamp` for individual message
* Fix `ConcurrentAggregatorTests` race condition.
Since currently the default release strategy is `SimpleSequenceSizeReleaseStrategy` which is just based on the `MessageGroup` size, there is no guaranty which messages will complete the group in concurrent environment.
The test is really based on the `SequenceAwareMessageGroup` logic to discard the message with the same `correlationId`
Fix `@Copyright` format
Move cast to `MessageHolder` after `Assert.isInstanceOf(MessageHolder.class, messageHolder)`
Retain backward compatibility in the `AbstractKeyValueMessageStore`
Polishing
JIRA: https://jira.spring.io/browse/INT-4137
Aggregators with `SequenceSizeReleaseStrategy` do not perform well with large groups
because of an O(n) linear search to reject (discard) duplicate sequences.
Change the aggregator to use a `SimpleSequenceSizeReleaseStrategy` by default, unless
`setReleasePartialSequences(true)`.
This avoids the linear search, which is not needed for most splitter -> ... -> aggregator
flows.
Log a warning if SSRS is used.
Polishing - PR Comments
* Remove `this.logger.isWarnEnabled()` since it is redundant when our log message is just string constant
JIRA: https://jira.spring.io/browse/INT-3825
Rework `MessagingMethodInvokerHelper` logic to accept non-public methods with Messaging annotations as candidates for invocation
Adjust SpEL configuration in the `MessagingMethodInvokerHelper` to deal with `declaredMethods()`, not only `public`
Honor caching in the `MethodReference`
JIRA: https://jira.spring.io/browse/INT-4131
Specify an expression on the outbound endpoints to set the `x-delay` header when
using the RabbitMQ Delayed Message Exchange plugin.
Polishing
- PR Comments
- Add `setDelay`
- Port `FunctionExpression` from DSL
- Add `SupplierExpression`
Javadoc Fixes
Use ValueExpression for delay
* Simple polishing for `SupplierExpression`
* Mention plain `delay` property in the `amqp.adoc`
JIRA: https://jira.spring.io/browse/INT-4132
The race condition is present when polling task may be ran before `MessageSource<?>` has been started.
* Swap the order of `start()` in the `SourcePollingChannelAdapter`.
Since proxying is now applied only for the `MessageSource.receive()` it doesn't hurt to start it before actual proxying.
Just because it is really should be started before performing its `receive()`
* Prove the proper order with the mock test and protect ourselves for the future similar changes
* Also swap the `stop()` order in the `SourcePollingChannelAdapter.
We have to stop/cancel the polling task before discarding internal `MessageSource` lifecycle.
For example with the current state we may close an underlying resource already, but still have the last polling tick.
That may cause any unexpected behaviour
**Cherry-pick to 4.3.x**
Polishing
