Fixes https://github.com/spring-projects/spring-integration/issues/8708
According to the `org.apache.sshd.common.channel.ChannelAsyncOutputStream.writeBuffer()` JavaDocs cannot be used concurrently.
* Introduce internal `DefaultSftpSessionFactory.ConcurrentSftpClient` extension
of the `DefaultSftpClient` to set a `Lock` around `super.send(cmd, buffer);`
* Remove lock from the `SftpSession` since it now is managed by the mentioned `ConcurrentSftpClient`
**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
# spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java
In the current code, an `IllegalStateException` might be thrown from the try block while invoking the `removeLockKeyInnerUnlink()` method, especially when caused by key expiration (resulting in `unlinkResult == false`).
This triggers the check block, which incorrectly sets the `unlinkAvailable` flag to `false`, even if the Redis server supports the unlink operation.
As a consequence, the subsequent `removeLockKeyInnerDelete()` method is invoked when it should not be.
* `IllegalStateException` should not be thrown from try block
* Add a comment and fix Checkstyle violations
**Cherry-pick to `6.1.x` & `6.0.x`**
(cherry picked from commit ba6d35d123)
Expected Behavior
Using a single Lua script to verify ownership of the lock and remove it.
Current Behavior
`unlock()` method of `RedisLock` uses two separate Redis operations:
* `isAcquiredInThisProcess()`` method executes a `GET` operation to verify if the lock is owned by the process.
* `removeLockKey()`` method executes `UNLINK/DEL` operation to remove the lock.
* The `removeLockKeyInnerUnlink()`, and `removeLockKeyInnerDelete()`
methods will execute a script both verify ownership of the lock and remove it.
**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
# spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
Fixes https://github.com/spring-projects/spring-integration/issues/8697
The `IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME` is not designed
for end-user code autowire, rather internal framework requirement
* Mark `BeanDefinition` for `IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME`
as `setAutowireCandidate(false)`
(cherry picked from commit f9dc75c739)
* 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)
Fixes https://github.com/spring-projects/spring-integration/issues/8674
The `ResourceKnownHostsServerKeyVerifier` does not take into account that several different
keys can be present in the known hosts resource for the same host/port
* Fix `ResourceKnownHostsServerKeyVerifier` to find a list of knows host for the requested session.
Then iterate of this result to match the key type first and then compare keys and their `revoked` marker
**Cherry-pick to `6.1.x` & `6.0.x`**
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`**
Fixes https://github.com/spring-projects/spring-integration/issues/8654
Spring Framework has an ability to start dependant beans automatically
when we start the current one.
The `AbstractMethodAnnotationPostProcessor` is missing a bean dependency
registration causing errors in target applications when Messaging Annotations configuration is used.
* Add `registerDependentBean()` into an `AbstractMethodAnnotationPostProcessor` when we generate
and register a `ConsumerEndpointFactoryBean`
* Change one of the `ClientManagerBackToBackTests` configuration to rely on a `@ServiceActivator`
for `Mqttv5PahoMessageHandler` bean to ensure that change in the `AbstractMethodAnnotationPostProcessor`
has a proper effect
**Cherry-pick to `6.1.x` & `6.0.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**
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`**
Fixes https://github.com/spring-projects/spring-integration/issues/8613
If `JsonPropertyAccessor` is registered for SpEL, it would be great
to have it working in native images as well.
Since SpEL is fully based on reflection, expose
`JsonPropertyAccessor$ComparableJsonNode` and `JsonPropertyAccessor$ArrayNodeAsList`
reflection hints for their method invocations from SpEL
**Cherry-pick to `6.0.x`**
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`
* Propagate Reactor context over headers
When we do something like `Flux.from(Publisher)`
and don't compose it with the one involved in the `Subscriber` context,
we lose this context.
* Provide a mechanism to propagate a Reactor context over message header
produce within that context.
* Restore this context in the `FluxMessageChannel` for a new publisher
we use in this channel
**Cherry-pick to `6.0.x`**
* Fix language in docs
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/8582
* Introduce a `PostgresSubscribableChannel.setTransactionManager()`
to wrap a message polling and dispatching operation into a transaction
* In addition add a `RetryTemplate` support around transaction attempts
**Cherry-pick to `6.0.x`**
* GH-8577: Deprecate ImapIdleCA.sendingTaskExecutor
Fixes https://github.com/spring-projects/spring-integration/issues/8577
When we process mail messages in async manner, it is possible that we end up
in a race condition situation where the next idle cycle closes the folder.
It is possible to reopen the folder, but feels better to block the current idle
cycle until we are done with the message and therefore keep folder opened.
* Deprecate `ImapIdleChannelAdapter.sendingTaskExecutor` in favor of an `ExecutorChannel`
as an output for this channel adapter or similar async hand-off downstream.
* Make a default one as a `SyncTaskExecutor` to make a sense of this deprecation
* Fix language in docs
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/8581
Do not overwrite configuration of externally provided `SshClient` in the `DefaultSftpSessionFactory`
* Replace JUnit `assertDoesNotThrow` by AssertJ `assertThatNoException` in test
**Cherry-pick to `6.0.x`**
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`**
* Some `LockRegistryLeaderInitiator` improvements
It is better to not go to the target lock provider at all
if the current thread is already interrupted.
* Check for the `Thread.currentThread().isInterrupted()` in the `while` loop
and `restartSelectorBecauseOfError()` immediately without checking for a lock
* Fix some other simple typos in the `LockRegistryLeaderInitiator`
**Cherry-pick to `6.0.x`**
* * Introduce a `LeaderSelector.yielding` flag
to revoke leader smoothly.
Turns out just canceling the `Future` may lead to a broken lock
where we cannot unlock it because the target lock repository may not work
with interrupted threads.
This way a new leader must wait until the lock is expired in the store
* Remove the use of instance suppliers on bean definitions
that are processed during the AOT phase.
* The remaining areas that use instance suppliers do so
at runtime and do not use reflection, but instead
are passed the configured bean to register.
See https://github.com/spring-cloud/spring-cloud-stream/issues/2655
**Cherry-pick to `6.0.x`**
* 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`
