There is no guarantee which message would reach the service first when we use `asyncConvertSendAndReceive()`.
* Add `config.longProcessEnteredLatch` to wait before sending the second (fourth in our case) message
until the long-running process against the first (third in our case) message has been started.
The timeout of `5 seconds` might not be enough on slow environment like GHA.
* Rework the test logic to rely on the `CountDownLatch` instead.
This way the `longer_process` would wait for it to be fulfilled,
while we count down only when another request has failed with timeout for locking.
Fixes: #9931
Issue link: https://github.com/spring-projects/spring-integration/issues/9931
If no message channel bean is declared explicitly, the XML parser for endpoints
register such a candidate into the global `ChannelInitializer`.
This way, the channel is created when this bean is initialized.
However, there are cases when endpoint bean could be called before `ChannelInitializer` initialization.
* Add `consumerEndpointBuilder.addDependsOn(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME);`
into the `AbstractConsumerEndpointParser` to ensure that `ChannelInitializer` bean is initialized
before the called endpoint bean.
* Adjust `ChannelInitializer` logic to use `registerSingleton()` API instead of `registerBeanDefinition()`
since the last one may cause problems with beans cache when this API is called at runtime.
**Auto-cherry-pick to `6.4.x`**
Fixes: #9890
Issue link: https://github.com/spring-projects/spring-integration/issues/9890
The `DestructionAwareBeanPostProcessor.requiresDestruction()` defaults to `true`.
As a result, beans that are not meant to be processed during the destroy lifecycle
are being registered for the `AbstractBeanFactory#registerDisposableBeanIfNecessary`.
for example, the `SimpleThreadScope` does not support destruction callbacks.
In other words, the `DestructionAwareBeanPostProcessor.requiresDestruction()` contract
is not merely for `DestructionAwareBeanPostProcessor` logic, but also has an effect
on the whole `BeanFactory`
* Implement `requiresDestruction()` in the `ControlBusCommandRegistry`
for same filter as `registerControlBusCommands()` does right now.
* Optimize behaviour using `MergedAnnotations` for the class instead of
`AnnotationUtils.findAnnotation()` twice, which creates the mentioned `MergedAnnotations` internally.
**Auto-cherry-pick to `6.4.x`**
Fixes: #9866
Issue link: https://github.com/spring-projects/spring-integration/issues/9866
Due to race condition between `sink.tryEmitNext()` and `sink.emitComplete()``,
there could be a situation when `onNext` signal slips after `onComplete`.
Appears on fast producers into this `FluxMessageChannel`.
That leads to error like `Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.`
* Fix `FluxMessageChannel` to check for `this.active` one more time just before `sink.tryEmitNext()` call
* Also mitigate a race condition with a `Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1))` in the `destroy()`
instead of `Sinks.EmitFailureHandler.FAIL_FAST`.
The `busyLooping()` would retry requested `onComplete()` signal for that specific error until success or timeout
**Auto-cherry-pick to `6.4.x` & `6.3.x`**
Issue link: https://github.com/spring-projects/spring-integration/issues/9863
The `PollerMetadata.sendTimeout` has been deprecated in previous versions
in favor of ``
* Remove `PollerMetadata.sendTimeout` and `PollerSpec.sendTimeout`
in favor of `SourcePollingChannelAdapterSpec.sendTimeout`
Fixes: #9863
Issue link: https://github.com/spring-projects/spring-integration/issues/9863
The `PollerMetadata.sendTimeout` has been migrated to the `SourcePollingChannelAdapter.setSendTimeout()` long time ago.
Right now this option is not propagated anywhere
* Introduce missed `SourcePollingChannelAdapterSpec.sendTimeout`
* Deprecate (for removal) `PollerMetadata.sendTimeout`
* Verify `SourcePollingChannelAdapterSpec.sendTimeout` option in the `ReactiveStreamsTests`
**Auto-cherry-pick to `6.4.x` & `6.3.x`**
Fixes: https://github.com/spring-projects/spring-integration/issues/9854
The well-known `Dispatcher has no subscribers` is not very informative
when a message is produced from early application context initialization phase
* Add internal `ApplicationRunningController` bean to handle early `start()` event
* Check for this bean status from the `AbstractMessageChannel.send()`
* Throw specific `MessageDispatchingException` to indicate that the message was produced from a wrong place
* Adjust `ApplicationEventListeningMessageProducer` logic for `ContextStoppedEvent` & `ContextClosedEvent`
to indicate that `AbstractMessageChannel` bean might not dispatch a message because the application context is not running
Fixes: #9825
Issue link: https://github.com/spring-projects/spring-integration/issues/9825
The `DelayerEndpointSpec` extends `ConsumerEndpointSpec` which has a `taskScheduler()` option.
However this is set only to the endpoint for this `MessageHandler`.
* Override `taskScheduler()` method on the `DelayerEndpointSpec` to set
the provided `TaskScheduler` to the `DelayHandler` as well
**Auto-cherry-pick to `6.4.x` & `6.3.x`**
Fixes: https://github.com/spring-projects/spring-integration/issues/9792
The `StreamTransformer` closes `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header value,
so this becomes unusable afterward.
In addition, it may even cause some problems downstream when the message could be serialized for
subsequent network interaction.
* Add logic to the `StreamTransformer` to build a new message, but remove `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header
* Verify that header is removed in the `StreamingInboundTests`.
Addition `StreamingInboundTests` clean up for better code style.
Fixes: https://github.com/spring-projects/spring-integration/issues/9754
Right now a correlation handler can discard messages in the expired group one by one.
In some scenarios it would be useful to have single message in discard for the whole group.
* Expose `discardIndividuallyOnExpiry` for the `AbstractCorrelatingMessageHandler`,
and `AggregatorFactoryBean`, and respective `CorrelationHandlerSpec` for DSL.
This new option takes action only if a `discardChannel` is provided,
and `sendPartialResultOnExpiry` is not set to `true`.
When `discardIndividuallyOnExpiry` is false, the messages in the expired group are packed
into a list for payload of a discarding single message.
* Test and document the new feature
Fixes: #9745
Issue link: https://github.com/spring-projects/spring-integration/issues/9745
The `IntegrationEvaluationContextFactoryBean` does not provide a `TypeLocator` by default.
That may lead to a class-not-found problem from different `ClassLoader` in the parallel Java `Stream` executor.
* Fix `IntegrationEvaluationContextFactoryBean` to use a `StandardTypeLocator` based on the `applicationContext.getClassLoader()`
**Auto-cherry-pick to `6.4.x` & `6.3.x`**
Fixes: https://github.com/spring-projects/spring-integration/issues/9743
Spring Integration provides observation for the `MessageChannel`, `MessageHandler`
and `MessageProducerSupport`.
The `SourcePollingChannelAdapter` is missing, and it is that only special endpoint which
deals with `MessageSource` implementations via scheduled tasks in the poller.
Essentially, this endpoint is a start of the flow, but it still is a consumer of data from the source system.
* Add an `Observation` logic to the `SourcePollingChannelAdapter`.
* Divide it into two phases: start (and open scope) when message is received; stop (and close scope) when the whole polling task for a message is done.
We need this separation because of transaction scope for the polling task.
At the same time we don't want to emit an observation for a void polling task.
* Change `MessageReceiverContext` to accept a `handlerType`.
The `MessageHandler` contributes a `handler`.
The new support in the `SourcePollingChannelAdapter` - `message-source`.
And change `MessageProducerSupport` to contribute a `message-producer`
* Verify the single trace is supported for the whole flow (including transaction synchronization) starting from a `SourcePollingChannelAdapter` in a new `SourcePollingChannelAdapterObservationTests`
* Document this new feature
Fixes: https://github.com/spring-projects/spring-integration/issues/5123
When `RedisMessageStore`, for example, adds and removes messages, it operates on two keys separately, which may cause problems in multi-threading due to non-atomic operations.
Although using Redis to delay messages is not a good idea, the abnormal loss of messages in the logs alerted me when the number of requests was not large.
By comparing the logs, the problem that the message group representing the metadata is not consistent with the actual message.
A simple solution is to add lock like in the `SimpleMessageStore`, which is also the approach taken in this pull request.
* Add `LockRegistry` to `AbstractMessageGroupStore`
* Normalize access levels and method name about the lock of `MessageGroupStore`
* Add document about the lock of `AbstractMessageGroupStore`
Fixes: https://github.com/spring-projects/spring-integration/issues/9706
Previously, if a `MessageGroupProcessor` returns a collection of payloads,
the `AbstractCorrelatingMessageHandler` has failed with the `IllegalArgumentException`
stating that only collection of messages is possible.
From now on such a restriction is eliminated and returned collection of payloads
is emitted as a single reply message from the aggregator.
* Add `AbstractCorrelatingMessageHandler.isResultCollectionOfMessages()`
to return `true` only if result is a collection of messages, treating them as a "partial sequence".
* Deprecate `AbstractCorrelatingMessageHandler.verifyResultCollectionConsistsOfMessages()`
since it is out of use now.
Fixes: https://github.com/spring-projects/spring-integration/issues/9683
Since we don't support SpEL-based Control Bus functionality anymore,
there is no need to keep separate `controlBusOnRegistry()`
* In all DSLs Deprecate `controlBusOnRegistry()` and restore `controlBus()` which is now fully based
on the `ControlBusCommandRegistry`
* Deprecate now out of use `<control-bus use-registry="">` attribute
* Fix `ControlBusParser` to not deal with `use-registry` attribute anymore
* Remove deprecated before `ExpressionControlBusFactoryBean` and `ExpressionCommandMessageProcessor`
* Remove `use-registry` from test configs
* Fix `ControlBusChainTests` to rely on a new Control Bus functionality
Fixes: #9709
Issue link: https://github.com/spring-projects/spring-integration/issues/9709
The Java DSL loses an `inputChannel` when existing channel is referenced by its name.
Then the target IntegrationFlow does not contain a bean reference for this channel and when we call its `getInputChannel()` we got something from middle of the flow.
However, that `inputChannel` is really expected to be an input for the flow.
* Fix `IntegrationFlowBeanPostProcessor` to get a bean by `MessageChannelReference` and populate it into `targetIntegrationComponents`
as we do for newly created `DirectChannel` if there is no bean for provided `MessageChannelReference`
**Auto-cherry-pick to `6.3.x`**
Fixes: #9705
Issue link: https://github.com/spring-projects/spring-integration/issues/9705
Currently when any class that implements `AbstractReplyProducingMessageHandler` doesn't produce a reply it will log the message, even if no reply is required.
The message should only be logged if `isLoggingEnabled()` returns true as a handler that doesn't require a reply may be a normal operation that is expected, and if we've set `loggingEnabled` to false it shouldn't log the message.
* Prevent logging when no reply is provided for an `AbstractReplyProducingMessageHandler` if logging is disabled
**Auto-cherry-pick to `6.3.x`**
Fixes: #9702
Issue link: https://github.com/spring-projects/spring-integration/issues/9702
The current `CheckedCallable.unchecked()` returns `Runnable`, which is not an expectation.
* Deprecate `CheckedCallable.unchecked()` in favor of newly introduced `CheckedCallable.uncheckedCallable()`.
We cannot call it `unchecked()` as well, since `Callable` after erasure becomes similar to class signature as `Runnable`.
**Auto-cherry-pick to `6.3.x`**
Related to: https://github.com/spring-projects/spring-integration/issues/9683.
Since we are going to restore `controlBus()` for convenience,
and only one way of Control Bus pattern, based on the `ControlBusCommandRegistry`, there is no need to have another `controlBusOnRegistry()` method
* Also fix link for `error-handling` chapter from the `special-channels.adoc`
Such a `Thread.sleep()` makes tests slower, plus may cause a longer timing issue on slow CI/CD when CPU resources are limited
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9623
Issue link: https://github.com/spring-projects/spring-integration/issues/9623
The `ConcurrentModificationException` is thrown from the `ThreadStatePropagationChannelInterceptor.MessageWithThreadState.stateQueue`
which is a not thread-safe `LinkedList`
* Fix `ThreadStatePropagationChannelInterceptor.MessageWithThreadState.stateQueue` to be a `LinkedBlockingQueue` instead
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9617
Issue link: https://github.com/spring-projects/spring-integration/issues/9617
The `ListenableFuture` is marked `forRemoval` in Spring Framework.
So, fix the code base to use `@SuppressWarnings("removal")`.
Also, add a warning into logs that `ListenableFuture` support will be removed in `7.0`.
* Fix JavaDocs where `ListenableFuture` is mentioned in favor of `CompletableFuture`
Fixes: #9614
Issue link: https://github.com/spring-projects/spring-integration/issues/9614
The `Matcher.quoteReplacement()` escapes `\` as well as `$`.
However, Jackson tries to resolve special symbol from the escaped `$` and fails as `Unrecognized character escape '$' (code 36)`
* Fix `SimpleJsonSerializer.toElement()` to escape only `\`
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9416
Issue link: https://github.com/spring-projects/spring-integration/issues/9416
The `MessageBuilderFactory` bean could be used a central place to provide custom `Message`
implementation into the application.
For example, the `GenericMessage.toString()` can be overridden to remove or mask sensitive
information from the payload or headers.
* Extract a `BaseMessageBuilder` from the `MessageBuilder` class to simplify
a custom `MessageBuilderFactory` implementation
* Test and document new feature and its purpose
Fixes: #9561
PR: https://github.com/spring-projects/spring-integration/pull/9561
The `DelayHandler.DelayedMessageWrapper` cannot be deserialized when using `RedisMessageStore` and JSON serialization:
```
org.springframework.data.redis.serializer.SerializationException: Could not read JSON:Cannot construct instance of `org.springframework.integration.handler.DelayHandler$DelayedMessageWrapper` (no Creators, like default constructor, exist): cannot deserialize from Object value (no delegate- or property-based Creator)
at [Source: UNKNOWN; byte offset: #UNKNOWN]
at org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer.deserialize(GenericJackson2JsonRedisSerializer.java:311)
```
* More code clean up and refactoring in the test
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Although `RenewableLockRegistry` provides a renew interface, it is inconvenient for users.
Developers hope to have a lock that can be automatically renewed.
On the one hand, it can avoid subsequent failures caused by locks that will not expire when abnormal exits,
and on the other hand, it can avoid unlock failures caused by lock expired.
* Add `RenewableLockRegistry.setRenewalTaskScheduler()` and when it is set, schedule a `renew()` script periodically
when lock is acquired from Redis with `1/3` of `expireAfter`
* Test and document the feature
Fixes: #9524
Issue link: https://github.com/spring-projects/spring-integration/issues/9524
It is useful in some use-cases to be able to inject a custom `TaskScheduler`
(e.g. with a `TaskDecorator`) into a source polling channel adapter.
* Add `SourcePollingChannelAdapterFactoryBean.setTaskScheduler()`
and call it from the `SourcePollingChannelAdapterSpec.taskScheduler()`
* Fix JavaDocs typos in the `ConsumerEndpointSpec`
* Test custom `TaskScheduler` usage and mention new option in the `whats-new.adoc`
