* Make `@IntegrationConverter` Native compatible
* Add `BASE_PACKAGE` into an `IntegrationContextUtils`;
deprecate similar in the `IntegrationConfigUtils`.
This fixes a package tangle between `config` and `context`
* Move `ConverterRegistrar` and `CustomConversionServiceFactoryBean`
into a `config` package since they are package protected and
created their instances in the `IntegrationConverterInitializer`
functional way instead of reflection
* Use new `IntegrationContextUtils.BASE_PACKAGE` constant instead of
deprecated one
* Make `DefaultConfiguringBeanFactoryPostProcessor` `public` to
make it available for Spring Native `trigger` option in the `@NativeHint`
declaration
* Simplify logic around `JsonPath` to just a `ClassUtils.isPresent()`
* Move the `@IntegrationConverter` processing logic into the `ConverterRegistrar`
to avoid reflection via `BeanDefinition` ctor arg manipulation
* Move the reflection logic into a `ConverterParser` which, being a part of XML
configuration, is not going to be compatible with native any way
* Mark `JsonNodeWrapperToJsonNodeConverter` with an `@IntegrationConverter`
since it is not registered via reflection any more
* Expose `MicrometerMetricsCaptorRegistrar.METER_REGISTRY_PRESENT` and use
it in the `IntegrationGraphServer`
* Extract `UnmarshallingTransformer.MIME_MESSAGE_PRESENT` for less
reflection at runtime
* Use `null` for a `ClassLoader` arg in the `ClassUtils.isPresent()`
relying on the default one internally
* * Fix Checkstyle violations
Currently the `MessageHistoryRegistrar` can parse several sources for message history -
`@EnableMessageHistory` and/or `<message-history>`.
Since its logic relies on the reflection it is not compatible with Spring Native.
Plus it causes confusion when several sources are declared so when time comes to
change something in that configuration, we may miss some place to re-align with
our new requirements.
Better to reject extra configurations and enforce end-users to use only one
`@EnableMessageHistory` or `<message-history>`.
This is actually a preferences for many other `@Enable...` in Spring portfolio.
Plus we got a benefit with a Spring Native compatibility
* Fix `MessageHistoryRegistrar` to parse only one `@EnableMessageHistory`.
Register `MessageHistoryConfigurer` function way for Spring Native compatibility
* Clean up `PublisherRegistrar` for better readability
* Fix message history tests which exposed several configurations
* Add JavaDoc into `EnableMessageHistory`
* Refactor `MessageHistoryConfigurer` to let to override patterns configuration
at runtime
* Clean up `message-history.adoc`
* Fix complexity in the `DefaultJmsHeaderMapper` according Sonar report
* Optimize all JMS tests to rely on a shared `ActiveMQConnectionFactory` resource
and disable JMX & statistics for embedded ActiveMQ broker
* Increase timeout for some sporadically failing tests
Related to: https://github.com/spring-cloud/stream-applications/issues/156
The `IntegrationReactiveUtils` uses a general `Flux.retry()`
operator to always retry for all the errors.
On the other hand it has only a `.doOnError(MessagingException.class)`
which leads swallowing all the other exceptions from logs and handling.
* Replace with `retryWhen()` for the `MessagingException` predicate
failing for all other exceptions.
The end-user may add their own `retry` or error handling mechanism to
the returned `Flux` from the `IntegrationReactiveUtils`
**Cherry-pick to 5.4.x**
* GH-3502: More refactoring to avoid reflection
Fixes https://github.com/spring-projects/spring-integration/issues/3502
* Move `ChannelInitializer` bean registration into an `AbstractIntegrationNamespaceHandler` -
it was never used for annotations and Java DSL...
* Rework `IntegrationFlows.fromSupplier()` to call a provided `Supplier` directly -
not via reflection in the `MethodInvokingMessageSource`
* Resolve new Sonar smells
* Rework `EndpointSpec` to accept an expected factory bean instance via ctor arg
instead of reflection
* Rework `Jackson2JsonObjectMapper` to use well-known module instances directly -
not via reflection from their class names
* * Revert `DefaultMethodInvokingMethodInterceptor.methodHandleCache` property definition wrap
* GH-3521: Delayer: schedule release task with TX
Fixes https://github.com/spring-projects/spring-integration/issues/3521
There is a race condition when transactional `MessageStore` is used
for `DelayHandler`, so the message is not visible for reads until after TX is
committed, but a scheduled release task may be already ready after delay
* Register a `TransactionSynchronization` with scheduling a releasing task
when TX is committed
**Cherry-pick to `5.4.x` & `5.3.x`**
* Fix language in delayer.adoc
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* INT-4377: aggregator groupTimeout as Date
JIRA: https://jira.spring.io/browse/INT-4377
Change the `groupTimeoutExpression` logic to let it to be evaluated to `Date`
instance for some fine-grained scheduling use-case, e.g. to determine a
scheduling moment from the group creation time (`timestamp`) instead of a
current message arrival
* Fix language in docs accoridng PR review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* INT-4444: Introduce `@Reactive` & `reactive()`
JIRA: https://jira.spring.io/browse/INT-4444
Right now the high-level API creates a `ReactiveStreamsConsumer`
only when the input channel is a `Publisher<?>` impl or target handler
is a `ReactiveMessageHandler`
* Add `@Reactive[] reactive()` attribute to messaging annotations
* Add `ConsumerEndpointSpec.reactive()`
Both options point to the same `ConsumerEndpointFactoryBean.setReactiveCustomizer()`
making the target endpoint always as a `ReactiveStreamsConsumer` independently of
the input channel and target handler
* Use the `Function` to customize a source `Flux` from the channel
* Test and document a new feature
* * Fix links in docs
* * Fix `ReactiveStreamsTests`
* * Rework `reactive()` attribute of messaging annotations ot a single `@Reactive` value
with default as `@Reactive(ValueConstants.DEFAULT_NONE)`
* Fix language in docs
* Fix `MessagingAnnotationUtils.resolveAttribute()` to use `requiredType.isInstance()`
instead of comparing classes since annotation instances are `Proxy` at runtime
* Miscellaneous improvement for end-user experience
Related to https://stackoverflow.com/questions/66428239/ineligiblemethodexception-found-more-than-one-parameter-type-candidate-spring
Related to https://stackoverflow.com/questions/66397350/spring-integration-how-do-i-call-the-spring-integration-from-spring-controller
The exception thrown from gateway method call for `null` arg is not clear that it cannot be `null`
* Improve `GatewayMethodInboundMessageMapper.mapArgumentsToMessage()` exception message to show args
and explain that it cannot be resolved to `null`
When more than one params are candidates for method invocation, the `MessagingMethodInvokerHelper`
throws an `IneligibleMethodException` which outcome is not clear what to do
* Add `@Payload` annotation recommendation into an exception message to make the reason more cleaner
* Some other code style clean up in the `MessagingMethodInvokerHelper`
* * Fix `PseudoTransactionalMessageSourceTests.testCommit()` for better coverage
* Apply suggestions from code review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
It looks like `OrderedAwareCopyOnWriteArraySetTests` uses
extra repetitions in its tests for nothing relevant to the
tested functionality.
Only the problem that it spawns too many threads
* Use a tmp file in the `FileTests.testFileReadingFlow()`
for writing content and then rename it to the target file.
It looks like we may already have a file just with a short content
and it is picked up by the watch service for processing in the flow
* Rework `AggregatorWithCustomReleaseStrategyTests` to JUnit 5 and remove
extra looping logic since it just does not add any extra coverage just
performance overhead via crating and destroying the same application ctx
* Rework `ChannelAdapterParserTests` to `@SpringJUnitConfig` managed test
JIRA: https://jira.spring.io/browse/INT-3585
Even if this fix doesn't provide an order support,
it does cover an original premise of an issue
With this fix the order of accessors doesn't matter.
* Catch an `AccessException` in the `JsonPropertyAccessor.canRead()` and return `false`
to let the rest accessors to deal with the property requested
Having a bean as a `java.util.Properties` is not confusing
and may lead to some conflicts in the real application.
Plus it is not so easy to configure: need to know all
the possible integration properties - bad end-user experience
* Make an `IntegrationProperties` as a public POJO
for easy configuration of global properties
* Deprecate a presence of the `java.util.Properties`
* Leave framework-created `integrationGlobalProperties` as
a `Properties` instance for backward compatibility
* Fix tests to expose an `IntegrationProperties` bean instead
of deprecated `Properties`
* Fix docs according the change and current recommendations
The logic in the `FluxMessageChannelTests.testFluxMessageChannelCleanUp()`
relies on the `finishLatch.countDown()` which happens before we
release the lock for a group in the aggregator.
So, when we call a `destroy()` as the next statement, we interrupt
the lock instead of unlocking.
This cause a race condition in the Reactor's `Sink` to terminate properly
* Expose `CorrelationHandlerSpec.releaseLockBeforeSend()` and set it to `true`
in the test to properly unlock before we call `finishLatch.countDown()`
* Deprecate a couple introduced before options in favor of their variants
with `set` prefix
* Fix new Sonar smells
* GH-3464: Treat 0 as special for polling endpoint
Fixes https://github.com/spring-projects/spring-integration/issues/3464
The `maxMessagePerPoll <= 0` is considered as an unbound `receive()` call.
End-users would like to have a special treatment for `0` value -
skip the `receive()` call altogether for the current polling cycle.
* Change the logic for a scheduled poller to not call `pollForMessage()`
and just log an INFO when `maxMessagePerPoll == 0`
* Fix reactive poller to deal with `maxMessagePerPoll == 0` properly
* Expose `maxMessagesPerPoll` as a `@ManagedAttribute` to let it to be
modified via Control Bus and JMX
* Test and document a new behavior
* * Fix unused imports in the test class
* Fix language in docs accoridng PR review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The current `Flux.take()` doesn't allow an arg `< 0` treating
it as an unbound request.
* Change `take()` to `limitRequest()` according strict `MessageSource.receive()`
producing expectations
* Treat `maxMessagesPerPoll < 0` as a `Long.MAX_VALUE` for unbound requests;
`0` is treated in the `limitRequest()` as "no more requests - cancel"
* Revise `AbstractPollingEndpoint` for `LogAccessor` usage
* Add `AbstractPollingEndpoint` class JavaDocs
* Fix tests according `AbstractPollingEndpoint` changes
**Cherry-pick to `5.4.x`**
* Expose `error.ignoreFailures` global property
The global default `errorChannel` can have more then only one logging subscriber.
There is no guarantee in which order subscribers are called and one of them fails
the rest are not called.
* Expose `spring.integration.channels.error.ignoreFailures` global property for
default `errorChannel` and make it `true` by default for better end-user experience
when it is expected to have error being handled by all the subscribers
* * Fix language in the `whats-new.adoc`
* * Fix `JmsTests` for error channel to propagate an exception
* More language fixes for whats-new.adoc
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
Fixes https://github.com/spring-projects/spring-integration/issues/3474
The `AbstractTransformer` implementation just wraps a thrown exception into
a `MessageTransformationException`.
It is expected with any method transformer impl (POJO, Lambda direct etc.)
the exception is thrown as is without extra wrapping into an `IllegalStateException`
for consistency
* Fix `LambdaMessageProcessor` to rethrow `RuntimeException` as is
and only wrap into an `IllegalStateException` for all other exceptions
* Fix tests for missed extra stack trace
* Copy request headers when reply is not a Message
Related to https://stackoverflow.com/questions/65887787/spring-integration-unexpected-behavior-on-error-handling-in-splitter-aggregator
Typically `AbstractMessageProducingHandler` implementation with the `shouldCopyRequestHeaders = false`
returns a `Message`(or `MessageBuilder`) as a reply, but when some request handler `Advice` is involved,
we may produce a plain payload and the reply message does not have request headers
* Improve `AbstractMessageProducingHandler` logic to copy request headers into a reply message builder,
when the original reply is not a `Message` or `MessageBuilder`.
This way we improve end-user experience and set them free from copying those headers manually
* Justify a new behavior with the `TransformerTests.testFailedTransformWithRequestHeadersCopy()`
* * Fix Checkstyle violations
Fixes https://github.com/spring-projects/spring-integration/issues/3446
* For better resources utilization provide a `Stream<Message<?>>` API
on the `MessageGroupStore`, `MessageGroup` and `MessageGroupQueue`
* Use this API in the `DelayHandler` when it reschedules persisted messages
* add `Comparable` contract to the initial wrapper so that expressions
containing selection/projection with JsonNode filtering are now possible
* render directly the value of a ValueNode. SpEL parser can now evaluate
correctly the object value when the expression deals with filtering on
`JsonNode` values (<, >, ==...)
* the 2 above changes make this possible:
`property.^[name.getTarget().asText() == 'value1'].name` can now be
written as simply as `property.^[name == 'value1'].name`. The expression
is now fully compatible with other PropertyAccessor types
* rename `ToStringFriendlyJsonNode` to `JsonNodeWrapper` as it reflects
more the capability of the new wrapper
* add a GenericConverter to be able to convert the `JsonNodeWrapper`
class into a `JsonNode` (or its derivative classes) directly while
calling `expression.getValue(..., JsonNode.class)`
* add the possibility to access list items in a json-path like index
=> a negative number will begin from the end of the list <=
* return null value when `JsonNode` cannot handle the property name
(or index)
* Change Json Converter
* Clean up the code style
* Convert JsonPropertyAccessor tests to JUnit 5
* GH-3450: Expose PubSub requireSubscribers option
Fixes https://github.com/spring-projects/spring-integration/issues/3450
* Add `PublishSubscribeChannel` ctors with the `requireSubscribers` option
with a direct delegation to the same option of underlying `BroadcastingDispatcher`
* Add factory methods to the `Channels` & `MessageChannels` to expose
this new `requireSubscribers` option
* Expose an XML `require-subscribers` attribute for the `<publish-subscribe-channel`
* Document this new `requireSubscribers` option
* Introduce a global `spring.integration.channels.error.requireSubscribers` property
for a default `errorChannel`
* Document this property and explain its default `true` in the `error-handling.adoc`
* Remove docs for `spring.integration.postProcessDynamicBeans` since it purpose was
removed since version `5.1` in favor of "always post-process behavior" as it was
always with all the `BeanPostProcessor`s
* * Fix language in docs according review
* * Add more docs about `requireSubscribers` and cross-links between chapters
* GH-3452: Warn Publisher.onError in NullChannel
Fixes https://github.com/spring-projects/spring-integration/issues/3452
The `NullChannel` can subscribe to the `Publisher` payload fully ignoring
the possible data according the `NullChannel` nature.
However an error thrown from the reactive stream processing is also ignored.
* Log WARN message from the `Subscriber.onError()` when `NullChannel`
subscribes to the `Publisher` produced to this channel.
* Mention the logic in the `NullChannel` docs; point to the
`ReactiveRequestHandlerAdvice` for further possible error handling in
the target application
**Cherry-pick to `5.4.x`**
* * Remove unused imports
* * Fix `PseudoTransactionalMessageSourceTests` for `LogAccessor`
from `NullChannel`
* Subscribe to Publisher in the NullChannel
If a payload of the message in the flow is a `org.reactivestreams.Publisher`,
it must be subscribed somewhere downstream to initiate reactive processing.
The `NullChannel` just ignores the message altogether and therefore `Publisher`
is lost
* Check the payload of the message in the `NullChannel` for `Publisher` type
and subscribe to it
* Verify in the test and mention this logic in the docs
**Cherry-pick to 5.4.x**
* Apply suggestions from code review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* Fix compatibility with upgraded dependencies
* Move `What's New` for `5.4` to `history`
* Add `XmlUnit` dependency since the latest AssertJ has deprecated its `isXmlEqualTo()`
* Remove `javax.annotation-api` and `apiguardian-api` as redundant dependencies
Related to https://build.spring.io/browse/INT-MASTERSPRING40-JOB1-1271
* Fix race condition with subscription in the `FluxMessageChannel`:
rely on the `doOnRequest()` to start producing from upstream publishers
* Remove `SourcePollingChannelAdapterTests` since
`Class.getDeclaredConstructor().newInstance()` doesn't rethrow an exception as is
but wrap it into the `InvocationTargetException`.
This is probably the main reason to deprecate a regular `Class.newInstance()`
* Use `LettuceConnectionFactory.setEagerInitialization(true)` in the `RedisAvailableRule`
to avoid possible dead lock with lazy init
* Adjust Redis tests for new `RedisAvailableRule` behavior
* Rely on `MProducerSupport.active` for `Flux`
* Fix `MessageProducerSupport` to extract an `active` flag and set it before
`isRunning` - the `Flux` subscription relies on the `takeWhile()`
where in case of `autoStartup = false` we will never start consume because
it is set to `true` already after `doStart()`
* Refactor all the `MessageProducerSupport` implementation with similar
`active` state to use already one from the super class
**Cherry-pick to 5.3.x**
* * Remove `MessageProducerSupport.setActive()`
to not let to mutate it from the implementations
* Set `active` to `false` in the `destroy()`
* Clean up and fix typos in the affected `JmsMessageDrivenEndpoint`
* * Pull `active` flag down to the `AbstractEndpoint`
* Set `active = true` in the `start()` before calling `doStart()`
* Do same for `active = false` in the `stop()`
* Clean up `AbstractEndpoint` impls to not call `doStart/doStop` for nothing
* Refactor endpoints to rely on the `active` state from the `AbstractEndpoint`
not their own
See SO for more info:
https://stackoverflow.com/questions/64456946/handle-exceptions-errors-other-than-messagingexception-ie-other-error-excepti
In the versions before `5.2.x` the `Error` was wrapped to the `MessagingException`
in the `AbstractRequestHandlerAdvice` and this one was handled properly
on the gateway level with its `errorChannel` configured
* Fix `MessagingGatewaySupport` to catch all the `Throwable` and try to send them as `ErrorMessage`
to the `errorChannel`.
Otherwise unwrap returned `MessagingException` and re-throw an `Error` as is to keep
the current behavior for non-handled exceptions
* Do not wrap `Error` into a `MessagingException` and re-throw as is if there is no `errorChannel`
**Cherry-pick to 5.3.x & 5.2.x**
With the `time-source` in Spring Cloud Stream we noticed
that first two items have the same (or close) value
when the `time-supplier` is used with the reactive output channel.
Turns out the `contactMap()` with `prefetch == 1` asks two
upstream elements immediately not waiting for a completion
for the fist one.
This way we were not able to update `triggerContext` for the proper
`lastCompletionTime`
* Change `AbstractPollingEndpoint` to `prefetch = 0` for the
"fair" upstream request
* Verify the behavior with the new
`ReactiveInboundChannelAdapterTests.testTimeSupplierConsistency()`
* Mark `ReactiveInboundChannelAdapterTests` as `@LongRunningTest`
since it is now pretty long waiting for all the 3 dates to verify
* Fix `InboundChannelAdapterAnnotationPostProcessor` to properly
register `MethodInvokingMessageSource` when we have more than one
`Supplier` with the `@InboundChannelAdapter` in the same configuration
class
If there is a `MeterRegistry` bean (any) in the application context,
we add a `MicrometerMetricsCaptor` bean which populates meters further
from the components.
In some case we may load a `MicrometerMetricsCaptor` bean too early
so, not all the stuff around `MeterRegistry` maybe ready.
See Spring Boot and its `MetricsAutoConfiguration`
* Fix the `MicrometerMetricsCaptorRegistrar` the way to rely on the
`ObjectProvider<MeterRegistry>` instead
* Add package protected ctor to the `MicrometerMetricsCaptor` to
provide a target `MeterRegistry` on demand
All of that will ensure that we use an already post-processed `MeterRegistry`
including Spring Boot auto-configuration
**Cherry-pick to 5.3.x & 5.2.x**
Related to https://github.com/spring-projects/spring-framework/issues/25884
* Don't use `MonoProcessor` & `FluxProcessor` since they are deprecated now
* Fix RSocket components to use an `AtomicReference` header instead of deprecated
`MonoProcessor`
* Introduce an internal `IntegrationRSocketPayloadReturnValueHandler` to handle
properly a Spring Integration case for inbound requests with its
`IntegrationRSocketEndpoint` implementation
* Don't do response handling in the `RSocketInboundGateway` any more
since it is deferred now to the `IntegrationRSocketPayloadReturnValueHandler`
* Remove internal `ChannelSendOperator` since it is out of use from now on
* GH-3386: Add method signature for gateway proxy
Fixes https://github.com/spring-projects/spring-integration/issues/3386
All the gateway proxy method invokers are supplied with the same
bean name inherited from the proxy.
* Add method signature for proxy method bean to fine-grain the
management for those bean in the logs, message history and metrics
* * Use `Class.getSimpleName()` instead to make gateway names less verbose
* GH-3376: Remove gauges on application ctx close
Fixes https://github.com/spring-projects/spring-integration/issues/3376
The `MeterRegistry` may request meters on application shutdown.
The gauges for channels, handlers and message sources don't make sense
at the moment since all those beans are going to be destroyed.
* Remove gauges for channel, handler and message source numbers from the
`IntegrationManagementConfigurer.destroy()`
**Cherry-pick to 5.3.x & 5.2.x**
* * Add `MicrometerImportSelector` to conditionally load
a `MicrometerMetricsCaptorConfiguration` when `MeterRegistry`
is on class path.
* Make `MicrometerMetricsCaptorConfiguration.integrationMicrometerMetricsCaptor()`
bean dependant on the `ObjectProvider<MeterRegistry>`
* Make `IntegrationManagementConfiguration.managementConfigurer()`
dependant on the `ObjectProvider<MetricsCaptor>`.
This way the `IntegrationManagementConfigurer` is destroyed before
`MeterRegistry` when application context is closed
* Deprecate `MicrometerMetricsCaptor.loadCaptor()` in favor of
`@Import(MicrometerImportSelector.class)`
* * Add `MicrometerMetricsCaptorRegistrar` to register a `MICROMETER_CAPTOR_NAME`
bean when `MeterRegistry` is on class path and no `MICROMETER_CAPTOR_NAME` bean yet.
* Make `IntegrationManagementConfiguration.managementConfigurer()`
dependant on the `ObjectProvider<MetricsCaptor>`.
This way the `IntegrationManagementConfigurer` is destroyed before
`MeterRegistry` when application context is closed
* Deprecate `MicrometerMetricsCaptor.loadCaptor()` in favor of
`@Import(MicrometerMetricsCaptorRegistrar.class)`
* Fix test to make a `MeterRegistry` bean as `static` since
`@EnableIntegrationManagement` depends on this bean definition now
* Fix unused import in the `IntegrationRequestMappingHandlerMapping`
* Fix deprecations from Reactor
* Fix race condition in the `AbstractCorrelatingMessageHandlerTests`:
the discard message is sent much earlier than group is removed from the store.
Iterate group count call until it pass or 10 seconds timeout
* Remove list size assert in the `FtpServerOutboundTests`: looks like
it is not updated properly even if we have an expected content in the collection
* Increase timeout to assert remote files removal in the `FtpRemoteFileTemplateTests`
* Use `tryEmitNext` on Reactor `Sink` since `emitNext` is deprecated
* Add `MessageDeliveryException` emission when `send()` returns `false`
in the `FluxMessageChannel` for `subscribeTo` provided `Publisher`
* GH-3374: Fix scan for BF propagation
Fixes https://github.com/spring-projects/spring-integration/issues/3374
An internal `ClassPathScanningCandidateComponentProvider` instance in the `IntegrationComponentScanRegistrar`
does not propagate a provided `registry`.
* Implement `getRegistry()` on the internal `ClassPathScanningCandidateComponentProvider` to propagate
a provided into the `registerBeanDefinitions()` a `BeanDefinitionRegistry`
* Add `@Conditional` on some scanned `@MessagingGateway` in the `EnableIntegrationTests`
**Cherry-pick to 5.3.x & 5.2.x**
* * Remove unused import
* Restore `unused` warning on the unused registry arg
* Implement `Emission` handling in the `FluxMessageChannel` and `IntegrationReactiveUtils`
* Upgrade to Spring Kafka `2.6.0`
* Fix R2DBC components for deprecation in the Spring Data R2DBC
* Implement `StatementMapper.SelectSpec` for query expression
* Clean up for some sporadic test failures
