Related to https://build.spring.io/browse/INT-MAIN-84/
The `lock.unlock()` must be called in the `finally` block of the
nested `try..catch`, not in the outer which may just fail on `lock.lockInterruptibly()`
in which case there is just not going to be anything we can `unlock()` in the end
**Cherry-pick to `5.4.x` & `5.3.x`**
* Add `org.springframework.integration.history` to trusted default packaged of the `JacksonJsonUtils`
* Add `@JsonCreator` to `MessageHistory` `private` ctor to let it to be created automatically by Jackson
* Add `equals()` and `hashCode()` into `MessageHistory` for the proper `Message` comparison
* Add `MessageHistory` into headers for testing with Redis JSON (de)serialization
**Cherry-pick to `5.4.x` & `5.3.x`**
The `spring.integration.properties` can come with the `noAutoStartup`
property where we can specify a source polling channel adapter endpoint to
not start automatically.
Turns out the `SourcePollingChannelAdapterFactoryBean` propagates its
`autoStartup` property unconditionally which will skip the `noAutoStartup` value
because an `AbstractEndpoint.setAutoStartup()` sets an `autoStartupSetExplicitly` state
* Fix `SourcePollingChannelAdapterFactoryBean` to rely on a `Boolean` object state
and don't call target endpoint `setAutoStartup()` if it was not set
* Adjust `spring.integration.properties` in tests to use `noAutoStartup`
for some `SourcePollingChannelAdapterFactoryBean`
* Verify that property was applied in the `IntegrationFlowTests.testWithSupplierMessageSourceImpliedPoller()`
**Cherry-pick to `5.4.x` & `5.3.x`**
The `@Indexed` comes without any attributes, but it is included
into a chain via `importingClassMetadata.getMetaAnnotationTypes(ann)`.
So, skip attributes from meta annotation when it is `null`
**Cherry-pick to `5.4.x` & `5.3.x`**
We fail with `NoClassDefFoundError` when we use `@EnableIntegrationGraphController`
in WebFlux env without Spring MVC.
Another issue that we don't register CORS for WebFlux
* Extract top-level package protected classes for MVC and WebFlux to register CORS
in the appropriate environment according classpath
* Adjust `HttpIntegrationConfigurationInitializer` and `WebFluxIntegrationConfigurationInitializer`
for native compatibility
* Add `@Indexed` for `@MessagingGateway` for indexer support in Spring Boot and Spring Native
**Cherry-pick to `5.4.x` & `5.3.x`**
# Conflicts:
# spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerRegistrar.java
Fixes https://github.com/spring-projects/spring-integration/issues/3558
Kotlin lambdas mostly used to configure endpoints in DSL manner
are not really Java lambdas, but rather anonymous classes implementing
respective Java interfaces.
While in most cases such classes carry generic info for their method impls
properly in Java, it is somehow doesn't work well for `GenericHandler`
implemented by Kotlin lambdas
* Wrap provided `GenericHandler` in the `BaseIntegrationFlowDefinition.handle()`
into a Java lambda and call `handle()` recursively to carry an expected type to
the `LambdaMessageProcessor`
* Fix `LambdaMessageProcessor` to handle `ClassUtils.isKotlinUnit()` result of
an invocation as a `null` reply
**Cherry-pick to `5.4.x` & `5.3.x`**
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-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>
Fixes https://github.com/spring-projects/spring-integration/issues/3497
The `BaseIntegrationFlowDefinition.REFERENCED_REPLY_PRODUCERS` is filled up
only with the `MessageProducer` instances.
Therefore no reason to calculate a hash code for every single bean passed
to the `ReplyProducerCleaner.requiresDestruction()`
* Check for the `MessageProducer` instance before passing the bean to the
`BaseIntegrationFlowDefinition.REFERENCED_REPLY_PRODUCERS.contains()`
* Check for the `MessageProducer` in the `ReplyProducerCleaner.postProcessBeforeDestruction()`,
too.
It can be called independently of the `requiresDestruction()` and will calculate a hash code
from the bean again for nothing
**Cherry-pick to 5.4.x & 5.3.x**
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`**
Fixes https://github.com/spring-projects/spring-integration/issues/3473
When `AbstractEndpoint.start()` and `AbstractEndpoint.isRunning()`
are called from different thread on `synchronized` methods,
we may end up with a dead lock: one thread waits for monitor on
`synchronized` and another waits for the `lifecycleLock`
* Change `AbstractEndpoint.isRunning()` to a plain `return this.running;`
- there is no reason in a lock around returning this `volatile` property state
**Cherry-pick to 5.4.x & 5.3.x**
* 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>
Related to https://build.spring.io/browse/INT-MASTER-2240
Turns out the `doOnRequest()` is still not enough to be sure
that subscriber is accepted into the `Publisher`.
Probably because `doOnRequest()` maybe really asked for the
subscription by itself for some logic
* Use `Mono` with `repeatWhenEmpty()` until
`this.sink.currentSubscriberCount() > 0`
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
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`**
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`**
* 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**
* Use 'onErrorContinue' to continue receiving messages from Stream
* Send an `ErrorMessage` to the provided `errorChannel` (if any)
* And `@Nullable` to some `MessageProducerSupport` API which definitely
may accept `null`
* Extract common `buildMessageFromRecord()` method in the `ReactiveRedisStreamMessageProducer`,
so all the headers from the stream `Record` are carried to the message independently of the
record state - normal send or error sending
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
Fixes https://github.com/spring-projects/spring-integration/issues/3306
To prevent API misuse (one can set ID after doing a registration which
will do nothing actually), it is safer to remove set* methods from
`StandardIntegrationFlowRegistration` object.
Providing some related refactoring like making fields 'final',
adjusting usages to keep correct behavior of the builder.
* INT-3045: Add in & out ZeroMq channel adapters
JIRA: https://jira.spring.io/browse/INT-3045
* Add `ZeroMqMessageHandler` to produce messages into one-way ZeroMq sockets
* Add `ZeroMqMessageProducer` to consumer messages from one-way ZeroMq sockets
* Add `ConvertingBytesMessageMapper` impl for the `BytesMessageMapper` to
delegate an actual conversion into the provided `MessageConverter`
* Add `ZeroMqHeaders` for message headers constants representing ZeroMq message attributes
* Fix `ZeroMqChannel` for the proper deferred `zeroMqProxy` evaluation
* Add more JavaDocs
* Fix `ZeroMqChannelTests.testPubSubBind()` to be sure that really all the
subscribed channels get the same message from the `PUB` socket
* * Fix typo in the `ConvertingBytesMessageMapper`
* Add `this` for `doOnError()` in the `ZeroMqChannel` & `ZeroMqMessageProducer`
* Change the bind logic in the `ZeroMqMessageProducer` to `port` and let it to
bind to random port.
The actual port is available later via `getBoundPort()`
* Introduce a `ZeroMqMessageProducer.receiveRaw()` to let received `ZMsg` to
be produce as a `payload`
* Add a logic into `ZeroMqMessageHandler` to treat `ZMsg` in the payload of
request message as is without any conversion
* Fix race condition in the `ZeroMqMessageProducer` to destroy `consumerScheduler`
when the main `Flux` is complete
* * Add Java DSL for ZeroMq components
* Extract `ReactiveMessageHandlerSpec` for `ReactiveMessageHandler` impls
* Add debug message into `EmbeddedJsonHeadersMessageMapper` when cannot `decodeNativeFormat()`
* Make `ReactiveMongoDbMessageHandlerSpec` extending `ReactiveMessageHandlerSpec`
* Make `ZeroMqProxy` `autoStartup` by default
* Add `ZeroMqDslTests` to cover all the Java DSL for ZeroMq
* Introduce a `MimeTypeSerializer` to serialize a `MimeType` into JSON as a plain string;
use it as extra serializer in the `JacksonJsonUtils.messagingAwareMapper()`
* Fix typo for the `AllowListTypeResolverBuilder` inner class
* * Add some docs
* Fix Checkstyle violations
* * More docs
* Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The `Mono.doOnSuccess()` is always called for completed `MonoSink`
even if the value is `null`
* Fix `IntegrationReactiveUtils.messageSourceToFlux()` to check a
message for `null` before calling `AckUtils.autoAck()`
* Add an `logger.error()` message when `doOnError()` is invoked for that `Mono`
**Cherry-pick to 5.3.x**
* 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
Fixes https://github.com/spring-projects/spring-integration/issues/3367
Introduce a `requestTimeout` and `triggerTimeout` for `BarrierMessageHandler`
For instance, if an HTTP request sends a message to the barrier,
it should time out after 1min if no trigger message is received.
If the trigger message then arrives late and the HTTP request is no longer waiting,
it shouldn't wait for 1min before discarding the request but do so immediately.
