* GH-8685: Re-fetch group after setting condition
Fixes https://github.com/spring-projects/spring-integration/issues/8685
The `AbstractCorrelatingMessageHandler` updates the group metadata in DB
not only for provided `condition`, but also a `lastModified` field.
A subsequent scheduling for group timeout takes the `lastModified`
to compare with the value in the store after re-fetching group in task.
This does not reflect reality since adding `condition` modifies the data in DB,
but in-memory state remains the same.
* Re-fetch a group from the store in the `AbstractCorrelatingMessageHandler.setGroupConditionIfAny()`.
* Verify expected behavior via new `ConfigurableMongoDbMessageGroupStoreTests.groupIsForceReleaseAfterTimeoutWhenGroupConditionIsSet()`
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
* * Fix Checkstyle violation in the test
Within Spring Integration's WebSocket support, a `ConcurrentWebSocketSessionDecorator`,
which buffers outbound messages if sending is slow,
is used to decorate all websocket sessions in `IntegrationWebSocketContainer`,
the standard entrypoint for using websockets with Integration.
* Expose a `ConcurrentWebSocketSessionDecorator.OverflowStrategy` option on the `IntegrationWebSocketContainer`
**Cherry-pick to `5.5.x`, `6.0.x` & `6.1.x`**
(cherry picked from commit be53593af9)
# Conflicts:
# spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java
The `ConsumerEndpointSpec` extensions for Kotlin
don't delegate to the provided `endpointFactoryBean`
* Introduce `KotlinConsumerEndpointSpec` extension for `ConsumerEndpointSpec`
with the proper delegation to the provided spec
* Use `KotlinConsumerEndpointSpec` in the Kotlin-specific `Spec` classes
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
The `NumberFormatException` flow control is costly
operation
* Use `Character.isDigit()` check iterating through property String
instead of `NumberFormatException` flow control
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
Resolves https://github.com/spring-projects/spring-integration/issues/8638
Previously, immediate failures (e.g. timeout getting metadata) were
only thrown as exceptions, and not sent to the failure channel, if present.
**cherry-pick to all supported branches**
# Conflicts:
# spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
The `FluxMessageChannel` can subscribe to any volatile `Publisher`.
For example, we can call Reactor Kafka `Sender.send()` for
input data and pass its result to the `FluxMessageChannel`
for on demand subscription.
These publishers are subscribed in the `FluxMessageChannel`
and their `Disposable` is stored in the internal `Disposable.Composite`
which currently only cleared on `destroy()`
* Extract `Disposable` from those internal `subscribe()` calls
into an `AtomicReference`.
* Use this `AtomicReference` in the `doOnTerminate()`
to remove from the `Disposable.Composite` and `dispose()`
when such a volatile `Publisher` is completed
**Cherry-pick to `6.0.x` & `5.5.x`**
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
Fixes https://github.com/spring-projects/spring-integration/issues/8609
* Changed to passed the self instance to `addNewConnection()` instead of argument's connection
in a `TcpConnectionInterceptorSupport` to compose a chain of interceptors from top to down.
This way the target `Sender` get the last interceptor in a chain as its connection.
**Cherry-pick to `6.0.x` & `5.5.x`**
* GH-8600: Fix WebSocket `removeRegistration`
Fixes https://github.com/spring-projects/spring-integration/issues/8600
When we register a dynamic WebSocket endpoint and use a `WebSocketHandlerDecoratorFactory`
such an endpoint is not removed on an `IntegrationFlow` destruction.
The actual `WebSocketHandler` is decorated, however we still use an initial one
for condition.
* Refactor `IntegrationWebSocketContainer` to expose a `protected` setter for the
`WebSocketHandler` which is called from the `ServerWebSocketContainer` after decoration
**Cherry-pick to `6.0.x` & `5.5.x`**
* Fix language in Javadocs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
* Fix NPEs in DSL Specs
The `BaseWsInboundGatewaySpec` and `TailAdapterSpec` don't override super methods
and when we call them we fail with NPE since `target` was not populated.
* Fix `BaseWsInboundGatewaySpec` and its inheritors to populate the `target`
from their ctors.
* Remove redundant methods from `BaseWsInboundGatewaySpec` hierarchy
* Propagate `AbstractWebServiceInboundGateway` properties directly to the target
from the `BaseWsInboundGatewaySpec` inheritors
* Override `MessageProducerSpec` methods in the `TailAdapterSpec`
to populate respective option into the `FileTailInboundChannelAdapterFactoryBean`
* Expose more missed options for producer endpoint in the `FileTailInboundChannelAdapterFactoryBean`
**Cherry-pick to `6.0.x` & `5.5.x`**
* * `acceptIfNotNull()` for `errorMessageStrategy` in the `FileTailInboundChannelAdapterFactoryBean`
* * Fix `Unmarshaller` population logic in the `MarshallingWsInboundGatewaySpec`
Fix race condition, when methods `RedisLockRegistry#expireUnusedOlderThan` and `RedisLockRegistry#obtain` are executed successively.
It's possible to delete the lock from `RedisLockRegistry#expireUnusedOlderThan` method, when lock is created but is not acquired (`RedisLock#getLockedAt = 0`)
It can lead to the situation, when `RedisLockRegistry#obtain` returns multiple locks with the same redis-key, which shouldn't happen at all.
* Skip locks from expiration when their `lockedAt == 0` - new, not acquired yet.
**Cherry-pick to `6.0.x` & `5.5.x`**
# Conflicts:
# spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java
* GH-8562: Fix streaming source for remote calls
Fixes https://github.com/spring-projects/spring-integration/issues/8562
The `AbstractRemoteFileStreamingMessageSource.doReceive()` takes files first from a `toBeReceived` queue.
When `AbstractRemoteFileStreamingMessageSource.remoteFileToMessage()` fails to fetch the file content
because of interim connection issue, we reset this file from a filter and rethrow an exception.
The next `receive()` call will just go ahead to the next entry in the `toBeReceived` queue, but the
file we have just failed for will be retried only on the next list call to the remove directory.
This essentially breaks a possible in-order target application logic.
* Introduce `AbstractRemoteFileStreamingMessageSource.strictOrder` option to clear the `toBeReceived` queue
when we fail in the `remoteFileToMessage()`, so the next `receive()`
call would re-fetch files from remote dir, because the filter has been reset for those files.
* Fix `AbstractFileInfo.toString()` to not perform remote calls when we just log this file.
For example, we reset the file for connection failure and log the message about it,
but it fails again because we request `size` of the file which may require a remote connection.
**Cherry-pick to `6.0.x` & `5.5.x`**
* * Revert `AbstractFileInfo` changes
* Override `toString()` in `SmbFileInfo` instead -
exactly the place where connection is used to obtain
file attributes like `size` or `lastModified`
Fixes https://github.com/spring-projects/spring-integration/issues/8550
Turns out the Paho MQTT client does not re-subscribe when connection re-established on automatic reconnection
* Fix `Mqttv5PahoMessageDrivenChannelAdapter` to always subscribe to its topics in the `connectComplete()` independently of the `reconnect` status
* Verify behavior with `MOSQUITTO_CONTAINER` image restart in Docker
Fixes https://github.com/spring-projects/spring-integration/issues/4002
Turns out the `ConsumerStopAction` was introduced in a point version to mitigate an unsubscription bug
and let to preserve a previous behaviour.
* Deprecate `ConsumerStopAction` in favor of just `cleanSession` flag in the `MqttConnectOptions`
**Cherry-pick to `5.5.x`**
The `TcpOutboundGateway` deals internally with a `CompletableFuture`.
However, the `AbstractMessageProducingHandler` cannot handle them until `6.0`.
* Use `CompletableToListenableFutureAdapter` to produce a reply from `TcpOutboundGateway`
which can be handled properly in the `AbstractMessageProducingHandler`
* GH-3993: Fix async race condition in TcpOutGateway
Fixes https://github.com/spring-projects/spring-integration/issues/3993
When `TcpOutboundGateway` is in an `async` mode and `CCF` is configured
not for `singleUse` an `Semaphore` around an obtained `TcpConnection` is involved.
If we fail on `TcpConnection.send()`, resources are not clean up, including the
mentioned `Semaphore`: in async mode this happens only when we receive a reply.
* Catch an exception on the `TcpConnection.send()` and perform `cleanUp()` in async mode.
* Add `cleanUp()` into a scheduled task from the `TcpOutboundGateway.AsyncReply`
when no reply arrives in time.
* Optimize `TcpOutboundGateway.AsyncReply` behavior to cancel no-reply scheduled task
when reply arrives into a `CompletableFuture`
**Cherry-pick to `5.5.x`**
* * Call `cleanUp()` from no response scheduled task
only if `future.completeExceptionally()` is `true`
Fixes https://github.com/spring-projects/spring-integration/issues/3986
The `ReactiveStreamsConsumer` in addition to plain `Subscriber` and `MessageHandler`
also supports a `ReactiveMessageHandler`, but `MockIntegrationContext` doesn't
handle a scenario when `ReactiveMessageHandler` is provided for consumer
* Rework the logic in the `MockIntegrationContext` to substitute a `ReactiveMessageHandler`
in the `ReactiveStreamsConsumer` which has a precedence in its logic over plain `Subscriber`.
* Wrap a plain `MessageHandler` mock into a `ReactiveMessageHandler` before substitution
in the `ReactiveStreamsConsumer`
* Reset `ReactiveStreamsConsumer.reactiveMessageHandler` with source `ReactiveMessageHandler`
or `null` respectively to an original `ReactiveStreamsConsumer` configuration
* Mention `ReactiveMessageHandler` use-case in the `testing.adoc`
**Cherry-pick to `5.5.x`**
* GH-3954: Fix WebFlux XML config for ambiguity
Fixes https://github.com/spring-projects/spring-integration/issues/3954
When `web-client` attribute is provided for the WebFlux outbound components configuration via XML,
the `encoding-mode` is silently ignored
* Check for `encoding-mode` attribute presence in the `WebFluxOutboundChannelAdapterParser`
when `web-client` is provided and throw respective exception to reject such a config
**Cherry-pick to `5.5.x`**
* * Improve error messages in the `WebFluxOutboundChannelAdapterParser`
* GH-3959: MqttConFailedEvent for normal disconnect
Fixes https://github.com/spring-projects/spring-integration/issues/3959
The `MqttCallback.disconnected(MqttDisconnectResponse)` in Paho v5 client is also
called when server initiates a disconnection.
In this case that `MqttDisconnectResponse` does not have a `cause` value
* Modify `MqttConnectionFailedEvent` to make a `cause` property optional
* Fix `Mqttv5PahoMessageDrivenChannelAdapter` & `Mqttv5PahoMessageHandler`
to not check for `cause`, but emit an `MqttConnectionFailedEvent` for any `disconnected()` calls
Unfortunately current Paho v3 client does not call `connectionLost()` for normal disconnections
and we cannot react for this callback with an `MqttConnectionFailedEvent`
**Cherry-pick to `5.5.x`**
* Fix language in doc
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/3955
`Mqttv5PahoMessageDrivenChannelAdapter` unsubscribes from all topics, even if `cleanStart/cleanSession` is set to `false`, thus not receiving offline messages after restart.
* unsubscribe `Mqttv5PahoMessageDrivenChannelAdapter` only when `cleanStart`
* add tests
**Cherry-pick to `5.5.x`**
# Conflicts:
# spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
* GH-3945: Fix `not eligible for getting processed`
Fixes https://github.com/spring-projects/spring-integration/issues/3945
The `IntegrationManagementConfiguration` produces an `IntegrationManagementConfigurer` which is a `BeanPostProcessor`.
According to Spring recommendation this kind of infrastructure beans must be declared as `static`.
Due to an `implements ImportAware, EnvironmentAware` nature of the `IntegrationManagementConfiguration`,
we cannot use `static @Bean` method.
But since the `IntegrationManagementConfiguration` is not involved in any bean post-processing,
it is safe to follow recommendation and mark it as a `@Role(BeanDefinition.ROLE_INFRASTRUCTURE)`.
* Fix `MessagePublishingInterceptor` to initialize `MessagingTemplate` and `DestinationResolver` lazily
* Fix `AbstractMethodAnnotationPostProcessor` to initialize `DestinationResolver` lazily
**Cherry-pick to `5.5.x`**
* Use `getChannelResolver()` internally in the `AbstractMethodAnnotationPostProcessor`
instead of direct property access which might not be initialized yet
* Use a plain `boolean` for `templateInitialized` in the `MessagePublishingInterceptor`
to avoid skips in other thread where and move on with still not initialized properties
* Remove unused import
* Fix `this.` prefix for `beanFactory` property reference
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
* GH-3938: Fix HTTP XML configuration for ambiguity
Fixes https://github.com/spring-projects/spring-integration/issues/3938
The `encoding-mode` is a property of the `RestTemplate`.
Therefore, it cannot be set on the component configuration together with
an externally injected `rest-template`
Even if `HttpRequestExecutingMessageHandler` has an assertion for such
an ambiguity, the XML parser just ignores this `encoding-mode`
when it encounters the `rest-template`
* Fix `HttpAdapterParsingUtils.verifyNoRestTemplateAttributes()` to check for not allowed
attributes with a `encoding-mode` as well
* Remove a `default` from the `encoding-mode` to not cause an ambiguity in the parser
* Fix some typos in the `spring-integration-http.xsd`
* Rework `OutboundResponseTypeTests` to JUnit 5
**Cherry-pick to `5.5.x`**
* * Fix error handling for `encoding-mode` in the `HttpAdapterParsingUtils`
* Cover `encoding-mode` and `rest-template` ambiguity with a test against failing XML configuration
* * Improve error message for ambiguous attributes in the `HttpAdapterParsingUtils`
* Fix double start for `AbstractEndpoint`
When we use POJO methods in the `IntegrationFlowAdapter`,
the `IntegrationFlowAdapter` is set as a `target` for the `MessagingMethodInvokerHelper`.
When endpoint is started by the application context, such a `start()` is propagated
down to the `MessagingMethodInvokerHelper`.
And in our case back into an `IntegrationFlowAdapter` instance.
This one, in turn, starts its `IntegrationFlow` internally which leads to the
start of the mentioned endpoint in the beginning.
Therefore, we cause a recursive `start()` call on this endpoint from itself.
The `running` flag is set when we are already done with the `doStart()` logic.
Therefore a recursive `start()` call leads to two concurrent polling tasks
in the `AbstractPollingEndpoint`.
* Check also for the `active` flag in the `AbstractEndpoint.start()`
and reset it in case of exception in the `doStart()`
**Cherry-pick to `5.5.x`**
* * Change assert for message timestamps to `isCloseTo()` with percentage
* * Change `catch` in the `AbstractEndpoint.start()` to `RuntimeException`:
the `doStart()` cannot throw unchecked exceptions by definition
* * Fix imports in `AbstractEndpoint`
