The `IntegrationManagementConfigurer` is a `BeanPostProcessor`
so it must not have direct dependency injection for other beans.
In our case it is a `MetricsCaptor` injected from the
`IntegrationManagementConfiguration`
* Fix `IntegrationManagementConfiguration` and `IntegrationManagementConfigurer`
to rely on the `ObjectProvider<MetricsCaptor>` instead
Tested against latest Spring Boot
**Cherry-pick to `5.3.x` & `5.2.x`**
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java
Fixes https://github.com/spring-projects/spring-integration/issues/3425
It turns out that `IntegrationManagementConfigurer.destroy()` is still too
late to remove `gauges` and Micrometer tries to gather them on application
context close, when many beans are already destroyed
* Catch `ContextClosedEvent` in the `IntegrationManagementConfigurer`
and removed `gauges` from there
* Remove `destroy()` impl since it is out of use already:
the `IntegrationManagementConfigurer` is not supposed to be in the
target application directly, so it looks safe to remove the `DisposableBean`
altogether
**Cherry-pick to `5.3.x` & `5.2.x`**
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java
The `InboundChannelAdapterAnnotationPostProcessor` doesn't use a bean method name
when it parses a `Supplier` bean and only uses a configuration class name + `get`
for method part
* Fix `InboundChannelAdapterAnnotationPostProcessor` to also include a bean method
name into the final bean name for the `MethodInvokingMessageSource` based on
the `Supplier` bean
* Modify `ReactiveInboundChannelAdapterTests` to add one more `Supplier` with the
`@InboundChannelAdapter` to ensure that configuration is still valid after the fix
**Cherry-pick to 5.2.x & 5.1.x**
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**
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**
* 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
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/EnableIntegrationManagement.java
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfiguration.java
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java
* Fix some deprecation warnings
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java
* 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
* Add gauges for queue channel size
The `QueueChannel` provides a current size and remaining capacity metrics
* Add Micrometer gauges into `QueueChannel` to expose the current values
of the size and remaining capacity
**Cherry-pick to 5.3.x, 5.2.x & 5.1.x**
* * Revert `@SuppressWarnings("unchecked")` for test
* Document new gauges for queue channel
* * Fix IntegrationManagementConfigurer for NPE on `metricsCaptor`
* Fix wording in meter descriptions
Co-authored-by: Michel Jung <michel.jung89@gmail.com>
Co-authored-by: Michel Jung <michel.jung89@gmail.com>
Related to https://stackoverflow.com/questions/62761903/spring-integration-reactive-streams-support-exception-in-creating-a-reactive
The `MessagingGatewaySupport` creates an internal endpoint
for consuming messages from the provided `replyChannel`.
The endpoint type depends on the channel type.
The `ReactiveStreamsConsumer` was missing a `BeanFactory` injection
causing an error when `ReactiveStreamsConsumer` tries to extract a default `ErrorHandler`
from `BeanFactory`
* Refactor `MessagingGatewaySupport` to inject a `BeanFactory` to all
the possible correlator endpoints.
Also always call `afterPropertiesSet()` for all of them
**Cherry-pick to 5.3.x & 5.2.x**
Close the pool so that any sessions returned after the factory is
`destroy()`ed are closed.
* Call `removeAllIdleItems()` in `close()`.
* Close sessions in `SftpStreamingMessageSourceTests`.
**cherry-pick to all supported branches**
* Fix MessagingGatewaySupport for reactive error
The `onErrorResume()` was in a wrong place for the
`doSendAndReceiveMessageReactive()`: we have to catch all the exceptions
from the top level `Mono`, not only a reply one as it was before.
Ensure in HTTP and WebFlux test that behavior is fixed
**Cherry-pick to `5.3.x` & `5.2.x`**
* * Remove unused imports
Co-authored-by: Artem Bilan <abilan@vmware.com>
Fixes https://github.com/spring-projects/spring-integration/issues/3276
The `onErrorResume` for the `MessagingGatewaySupport.doSendAndReceiveMessageReactive()`
was in wrong place: only for the `buildReplyMono` which works only
when an outbound flow is fully based on reactive channels.
With a regular direct channel we can get an exception from the
`sendMessageForReactiveFlow` which is not covered with the mentioned
`onErrorResume` for the error handling on the configured `errorChannel`
Cherry-pick to `5.2.x & 5.1.x`
Fixes https://github.com/spring-projects/spring-integration/issues/3266
It turns out that `AnnotationGatewayProxyFactoryBean.onInit()` implementation
parses a `@MessagingGateway` attributes ignoring possible properties
population by setters.
This way a Java DSL `GatewayProxySpec` becomes useless since all
its options are overridden by default values from a synthesized
`@MessagingGateway`.
Also it is inconsistency when we declare an `AnnotationGatewayProxyFactoryBean`
as regular bean, but then called setters are ignored
* Add `protected` getters into `GatewayProxyFactoryBean` for all
the properties which can be overridden by annotation attributes
* Fix `AnnotationGatewayProxyFactoryBean` to consult with those getters
before populating a property with value from the annotation
**Cherry-pick to 5.2.x**
Fixes https://github.com/spring-projects/spring-integration/issues/3253
The `IntegrationFlowBeanPostProcessor` uses a `containsBean()`
and then `getBeanDefinition()` to be sure that we don't override already
existing bean even if it is created in the parent(s) context.
The problem that `containsBean()` check the hierarchy, but `getBeanDefinition()`
doesn't.
So, we fail with `NoSuchBeanDefinitionException` if bean exists in the parent ctx
* Introduce an utility `IntegrationContextUtils.getBeanDefinition()` to
scan `BeanFactory` recursively for `BeanDefinition` for the requested `name`
* Use this tool in the `IntegrationFlowBeanPostProcessor` logic
**Cherry-pick to 5.2.x & 5.1.x**
Resolves https://github.com/spring-projects/spring-integration/issues/3230
Previously, methods with the same name as `Lifecycle` methods were unconditionally ignored.
- Use logic similar to `ControlBusMethodFilter` to explicitly compare the candidate Method
to those on `Paused` and `Lifecycle` and only ignore those.
- Explicitly disallow these methods when named - previously the check was only performed
if no method name was supplied.
* - reinstate `Pausable/Lifecycle` methods if explicitly requested.
- log `Pausable` methods that are not candidates.
* - Fix typo in test method.
**Cherry-pick to 5.2.x**
* Fix JsonToObjectTransformer for ClassNotFoundEx
Related to https://github.com/spring-projects/spring-integration/issues/3223
The `JsonToObjectTransformer` consults `JsonHeaders` first and tries to build
a `ResolvableType` from other type headers which may be just a string identifications.
In this case a `ClassNotFoundException` could be thrown if a `ResolvableType` cannot
be build against non-class identificators
* Add `valueTypeExpression` option to the `JsonToObjectTransformer` to let to
build a `ResolvableType` using any possible custom logic, e.g. resolving
target classes from some registry using their ids from the mentioned headers
**Cherry-pick to 5.2.x**
* * Fix English language mistakes
* Upgrade to Spring Data Moore-SR6
* Fix `TransactionHandleMessageAdvice` and `TransactionInterceptorBuilder`
for deprecations in the `TransactionInterceptor`
* Fix `TransactionHandleMessageAdvice` and `TransactionInterceptorBuilder`
consumers to expose new `TransactionManager`-based options and
deprecate `PlatformTransactionManager`-based
* Fix failing JMS tests to reuse an ActiveMQ Connection Factory with a
`trustedPackaged(*)`
**Cherry-pick to master**
Fixes https://github.com/spring-projects/spring-integration/issues/3223
The `json__TypeId__` header may have a value which cannot be resolved to the `Class` in the current classpath
So, skip `ResolvableType` building logic until we really sure that end-user wants to map JSON headers
* WARN a build exception that we can't load a class for the `json__TypeId__` when we try to build a `ResolvableType`
in the `DefaultAmqpHeaderMapper`
* Document the negation feature for JSON headers
**Cherry-pick to 5.2.x**
* Populate a JsonHeaders.RESOLVABLE_TYPE on reply
Fixes https://github.com/spring-projects/spring-integration-samples/issues/277
In the `AbstractAmqpOutboundEndpoint` the `JsonHeaders.RESOLVABLE_TYPE`
from request message is copied to reply message making inconsistency downstream.
The `JsonToObjectTransformer` consults first a `JsonHeaders.RESOLVABLE_TYPE`
and deserialize payload to wrong type
* Fix `DefaultAmqpHeaderMapper` to populate a `JsonHeaders.RESOLVABLE_TYPE`
alongside with other `JsonHeaders` populated from the reply AMQP message.
This way a `JsonHeaders.RESOLVABLE_TYPE` from request message won't have effect
* To get access to classes, supply `AbstractHeaderMapper` with a bean factory `ClassLoader`
* Introduce a couple utility methods into `JsonHeaders` for building a `ResolvableType`
**Cherry-pick to 5.2.x**
* * Fix code formatting for arguments wrapping
Fixes https://github.com/spring-projects/spring-integration/issues/3152
The upstream `gatherResultChannel` header has been missed when we produced a reply from nested scatter-gather
Added Test Case for Nested Scatter Gather test
Simplified the the test cases and added author in changed cases
Corrected codestyle issue in Travis CI
Removed additional OriginalReplyChannel and originalErrorChannel in Headers.
Added additional not to be executed line of code in test case.
Restored OriginalErrorChannel Header and removed error handling related fixes
* Clean up code style and improve readability
**Cherry-pick to 5.1.x & master**
Resolves https://github.com/spring-projects/spring-amqp/issues/1142
Resize down does not work when not enough already allocated to match the reduction.
Improved SimplePool test cases and fixed code style
Fixed code style
GH-3143: Test Polishing
- use preferred assertJ expected exception checking
- add a test for reducing a partially allocated pool
- check active counts
To ensure advice-chain applicability the provided plain
`MessageHandler` is wrapped into a `ReplyProducingMessageHandlerWrapper`
with its particular `.wrapper` bean name.
When we build a graph for integration components, we use a
`componentName` from the `IntegrationObjectSupport` to represent a
node for endpoint in the graph.
In most cases the component name is an endpoint id around a `MessageHandler`
* Populate the missed component name in a `ReplyProducingMessageHandlerWrapper`
in the `AbstractMethodAnnotationPostProcessor` so nodes in the graph has
a proper name for their endpoint in the application context
* Use EmitterProcessor in the FluxMessageChannel
The `EmitterProcessor` has a good logic to block upstream producer
when its downstream subscriber cannot keep up with overproducing.
* Rework `FluxMessageChannel` logic to rely on the `EmitterProcessor`
instead of `Flux.create()`
* Cancel `FluxMessageChannel` internal subscriptions in the `destroy()`
* Fix `ReactiveStreamsTests.testFluxTransform()` for the splitter's
delimiter
* Ensure in the `FluxMessageChannelTests.testFluxMessageChannel`
that we can have several concurrent subscribers to the
`FluxMessageChannel`
* * Use `flux.onComplete()` instead of iteration over subscribers
* Change `subscribers` list into just `AtomicInteger` count marker
* fix `DefaultSplitterTests` according a new logic in the `FluxMessageChannel`
* GH-3107: Add errorOnTimeout for TcpInboundGateway
Fixes https://github.com/spring-projects/spring-integration/issues/3107
The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout
* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option
* * Use `delaySubscription()` for subscribing publishers in the `FluxMessageChannel`
to wait until this one subscribed.
* Use an `EmitterProcessor` to catch subscriptions and pass them as a
signal to delayed upstream publishers
* Fix `FluxMessageChannelTests.testFluxMessageChannelCleanUp` to
verify an actual property instead of removed.
* Fix `RSocketOutboundGatewayIntegrationTests` for the proper subscription
into a `FluxMessageChannel` before actual interaction with an RSocket
gateway.
This should help us also to avoid some race conditions in the future
* Revert "GH-3107: Add errorOnTimeout for TcpInboundGateway"
This reverts commit fa6119ddc4e423e8b15ca3fdc9fc4c79ea9d56af.
* * Refactor `FluxMessageChannel` to use `ReplayProcessor` for `subscribedSignal`.
This one is used `delaySubscription` for the upstream publishers
* Use a `AtomicBoolean` for subscription state since `doOnSubscribe()`
is called before `EmitterProcessor` adds subscribers for its `downstreams`
* Use `publishOn(Schedulers.boundedElastic())` for upstream publishers
to avoid blocking over there when our `EmitterProcessor` doesn't have
enough demand
* Refactor reactive tests to have a subscription into the `FluxMessageChannel`
earlier than emission happens for it
* * Use `Flux.subscribe(Consumer)` instead of `doOnNext(Consumer).subscribe()`
* * Emit `subscribedSignal` value after `.subscribe(subscriber)`
instead of `doOnSubscribe`
* Check for `this.processor.hasDownstreams()` before emitting such an event
* * Use `this.processor.hasDownstreams()` as a value to emit for `subscribedSignal`.
This way we are less vulnerable race conditions when subscribers are changed
actively
Fixes https://github.com/spring-projects/spring-integration/issues/3114
The contract of SpEL with its
`getValue(EvaluationContext context, @Nullable Object rootObject)` is
that we need to deal with provided `rootObject` even if it is `null`
and don't consult with `context.getRootObject()`
* Fix `ExpressionEvalMap` to have an internal `rootExplicitlySet`
to indicate that `root` explicitly provided by consumer, even if it is null.
According this flag call respective `Expression.getValue()`
* Add `@Nullable` to methods and their arguments into `ExpressionEvalMap`
& `FunctionExpression` to honor `Expression` contracts
* Populate an `HttpEntity` explicitly into `ExpressionEvalMap` from the
`HttpRequestHandlingEndpointSupport` and `WebFluxInboundEndpoint` for
full picture
* Fix JavaDocs indentations in the `ExpressionEvalMap`
* INT-3494: Resolve dir for writing as a Resource
JIRA: https://jira.spring.io/browse/INT-3494
The expression for local directory can be resolved into a `Resource`
or resource location.
* Fix `ExpressionUtils.expressionToFile()` to support `Resource` and
also use `ResourceUtils.getFile(path)` when expression result is a string
* Modify tests to ensure that resource is resolved properly
* Upgrade affected tests to JUnit 5
* Mention an new functionality in docs
* * Improve Java doc for `ExpressionUtils.expressionToFile()`
* Finish the sentence in the `file.adoc`
Fixes https://github.com/spring-projects/spring-integration/issues/3107
The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout
* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option
* Add a setter for MessagingGatewaySupport.errorOnTimeout option
* Expose an `errorOnTimeout` option on the DSL's `MessagingGatewaySpec`
making all the out-of-the-box inbound gateways possible to react to the
`MessageTimeoutException` when no reply during reply timeout
* Propagate properly `errorOnTimeout` in the `JmsInboundGateway`
* Modify docs respectively
* Improve docs about `errorOnTimeout`
Related to https://github.com/spring-projects/spring-framework/issues/23812
* Modify `FunctionsTests.kt` test to be sure that fall back to SpEL
invocation in case of Kotlin lambda works as expected.
This is like an additional test to the fix in Spring Framework
* Currently `MessagingMethodInvokerHelper.getTargetClass()` logs a
debug message when it can't cast a `targetObject` into an `Advised`.
Such a message is misleading and really unnecessary noise in the target
applications
* Fix the logic of the method to check `targetObject instanceof Advised`
before trying to extract interfaces from proxy.
This way we don't need to catch any exceptions at all.
If something will fail, it should be bubbled upstream for possible fixes
* Convert `MethodInvokingMessageProcessorTests` to JUnit 5 and restore
a couple tests
Fix a logic in the `JsonToObjectTransformer.obtainResolvableTypeFromHeadersIfAny`
when we have both `JsonHeaders.RESOLVABLE_TYPE` and `JsonHeaders.TYPE_ID`
and `JsonHeaders.RESOLVABLE_TYPE` header is valid `ResolvableType`, so
`JsonHeaders.TYPE_ID` doesn't override its value to return
* GH-3096: Skip RESOLVABLE_TYPE header in mapping
Fixes https://github.com/spring-projects/spring-integration/issues/3096
When we sent an AMQP message we should not map a
`JsonHeaders.RESOLVABLE_TYPE` header which is a `ResolvableType` and
isn not compatible after converting to string
Also improve `JsonToObjectTransformer` to ignore a
`JsonHeaders.RESOLVABLE_TYPE` when it is type of String
* * Fix `obtainResolvableTypeFromHeadersIfAny()` logic
* Use `EmitterProcessor` for Channels adaptation
Related https://github.com/spring-cloud/spring-cloud-stream/issues/1835
To honor a back-pressure after `MessageChannel` adaptation it is better
to use an `EmitterProcessor.create(1)` instead of `Flux.create()`.
This way whenever an emitter buffer is full, we block upstream producer
and don't allow it to produce more messages
**Cherry-pick to 5.1.x**
* * Wrap every new subscription into a `Flux.defer()`
* Fix `ReactiveStreamsConsumerTests` to use a new `Subscription` after
each `stop()/start()` on the `ReactiveStreamsConsumer`
* * Remove unused imports
* Extract `BaseIntegrationFlowDefinition`
Kotlin inline reified functions cannot override as syntax sugar generic
Java methods.
* Extract a `BaseIntegrationFlowDefinition` with strong typed methods,
plus those which doesn't have a syntax sugar variants
* Leave in the `IntegrationFlowDefinition` only those generic methods
which are just a syntax sugar without any types requirements
This way we can write Kotlin inline reified functions as an extension
with the same names to make a Kotlin DSL much cleaner
* * Override methods in the IntegrationFlowDefinition for bytecode compatibility
* Extract `protected` getters and setters for `BaseIntegrationFlowDefinition`
properties to allow a proper extension
* Use those getters and setters in the code
* * Add // NOSONAR for overridden methods
