Fixes: #8983
Since collections in MongoDB cannot be deserialized to custom types anymore,
add a new `DocumentToMessageHistoryConverter` to convert the `Document` with `components`
into a `MessageHistory` instance back
(cherry picked from commit 809d1396af)
Fixes: #8967
In the `fair` mode the `StandardRotationPolicy` re-configures an `AbstractFetchLimitingMessageSource`
for a new directory (and possible new `ConnectionFactory`) in the `beforeReceive()`.
However, with default `maxFetchSize` (or bigger than `1`), the `receive()`` would poll `toBeReceived` internal queue
for files cached from the previous polling cycle.
Since we rotate the source immediately to a new set of options, all those cached files make no sense
or even can cause the problem on fetching their content in case of `AbstractRemoteFileStreamingMessageSource`
when we rotate to a new `ConnectionFactory`.
* Call `fetchLimitingMessageSource.setMaxFetchSize(1);` in the `StandardRotationPolicy.beforeReceive()`
when `fair && !this.initialized`
**Auto-cherry-pick to `6.1.x`**
Fixes: #8950
The `IntegrationFlow.toReactivePublisher(true)` makes `PublisherIntegrationFlow`
to be stopped from the beginning and waiting for the `Publisher.subscribe()`.
However only `EndpointSpec` components are marked as `autoStartup(false)`.
The `MessageProducerSupport` is not included, therefore unexpected messages
are produced to the channel in the end of flow without subscribers.
* Check for `AbstractEndpoint` component type in the `PublisherIntegrationFlow`
and mark it as `setAutoStartup(false)`
* Ensure in a new `ReactiveStreamsTests.messageProducerIsNotStartedAutomatically()` test
that `MessageProducerSupport` is not started from the beginning,
but rather when we subscribe to the `Publisher` via `StepVerifier`
(cherry picked from commit 2b576c57d6)
There is no need to post-process the framework internal BPPs.
Therefore, mark `GatewayProxyInstantiationPostProcessor`, `GlobalChannelInterceptorProcessor`, `IntegrationFlowBeanPostProcessor`
and `BaseIntegrationFlowDefinition.ReplyProducerCleaner` as an `AopInfrastructureBean`.
This way they are not post-processed via AOP `BeanPostProcessor`
* Migrate `org.springframework.integration.aop` tests to JUnit 5
* Fix race condition in the `FluxMessageChannelTests.noMemoryLeakInFluxMessageChannelForVolatilePublishers()`
where `FluxMessageChannel.upstreamSubscriptions` is updated from a different thread
(cherry picked from commit a8ed1dd9aa)
Fixes: #8894
* Guard `this.messageStore` access in the `DelayHandler` with existing `this.lock`
to avoid `ConcurrentModificationException`
**Cherry-pick to `6.2.x`**
(cherry picked from commit 4b43969438)
Fixes: #8889
* Gather the current thread local state for the message into `IntegrationMessageHeaderAccessor.REACTOR_CONTEXT` header
before `this.sink.tryEmitNext(messageToEmit)`
* Restore thread locals for the current message in the `ReactiveStreamsConsumer.SubscriberDecorator` just before `this.delegate.onNext(messageToDeliver)`
**Cherry-pick to `6.2.x`**
(cherry picked from commit dbd46dfe3d)
Fixes: #8879
To work around the problem with `$share/` subscriptions the `Mqttv5PahoMessageDrivenChannelAdapter` must provide a `subscriptionIdentifier` into `MqttProperties` on `subscribe()`
* Introduce a `Mqttv5PahoMessageDrivenChannelAdapter.subscriptionIdentifierCounter` according to the MQTT specification:
> 3.8.2.1.2 Subscription Identifier: [..]The Subscription Identifier is associated with any subscription created or modified as the result of this SUBSCRIBE packet. If there is a Subscription Identifier, it is stored with the subscription.
This one is associated with the MQTT session for the current subscriber and does not interfere into other sessions even if identifier is same from the counter.
It works because the Subscription identifier is per session and because you cannot have multiple connection with the same client ID.
**Cherry-pick to `6.2.x` & `6.1.x`**
# Conflicts:
# spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
Fixes: #8873
The `mqttClient.subscribe()` API does not check if properties are provided and fails on the
`subscriptionProperties.getSubscriptionIdentifiers().get(0)` call with an `IndexOutOfBoundsException`
* Use another `mqttClient.subscribe()` API in the `Mqttv5PahoMessageDrivenChannelAdapter` where there is not such a check
* Ensure that `addTopic(NAME)` works as expected in the `Mqttv5BackToBackTests`
**Cherry-pick to `6.2.x` & `6.1.x`**
# Conflicts:
# spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
Fixes: #8869
Do not set the `redisMessageListenerContainer`'s executor in the setter method
since the Redis Message Listener container is not initialised during configuration due to lazy loading
**Cherry-pick to `6.2.x` & `6.0.x`**
(cherry picked from commit 62fd3e6c20)
Due to an antora ordering issue, the reference docs on the latest GA from
the project site throws an HTTP 404.
See spring-projects/spring-framework#32083 for more details.
Related to: #8856
Many tests create their own `ThreadPoolTaskScheduler` beans.
Therefore, its default phase might affect the memory and performance.
* Use `phase = SmartLifecycle.DEFAULT_PHASE / 2` for manual
`ThreadPoolTaskScheduler` beans
* Migrate affected tests classes to JUnit 5
* Make some other configuration adjustments for better performance
**Cherry-pick to `6.2.x`**
(cherry picked from commit 39c99c0719)
Fixes: #8856
The `ThreadPoolTaskScheduler` in Spring Framework manages now a lifecycle
with a `Integer.MAX_VALUE` phase by default.
This causes a wait block for scheduled tasks in the `AbstractPollingEndpoint` instances.
These tasks are cancelled by in the later `Integer.MAX_VALUE / 2` phase.
So, we have a lifecycle deadlock on context close
* Fix `DefaultConfiguringBeanFactoryPostProcessor.registerTaskScheduler()`
with a `.addPropertyValue("phase", SmartLifecycle.DEFAULT_PHASE / 2)`
to let the `AbstractPollingEndpoint` to cancel its currently scheduled task
to avoid possible pollution with unexpected (and lost) messages
**Cherry-pick to `6.2.x`**
(cherry picked from commit 11e0ea1305)
Fixes: #8852
The `while (true) {` causes the thread to be blocked forever.
The `stop()` waits on the `if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {` the whole time
* Fix with the `while (isActive()) {`
* Remove the `ConnectionFactoryTests.testEarlyCloseNet()` since we cannot reach expectations because of
race condition between `start()` and `stop()`
**Cherry-pick to `6.2.x` & `6.1.x`**
(cherry picked from commit 0b60dfc9d1)
