The `ReactorNettyTcpStompClient` is based on the `ReactorNettyTcpClient`
which uses a `Schedulers.newParallel()` which has to be shut downed in the end
* Migrate all the STOMP tests to JUnit 5
* Remove redundant `@SuppressWarnings("deprecation")`
* Add `Duration.ofSeconds(10)` to `StepVerifier.verify()`
to avoid infinite wait and lose failing text context on the hang CI build
Fixes https://github.com/spring-projects/spring-integration/issues/3226
* Implement a `ReactiveRedisStreamMessageProducer` to consume Redis streams
* Add support for consumer groups and group auto-creation
* Add `@author` to `RedisHeaders`
* Rename test methods in the `ReactiveRedisStreamMessageHandlerTests` to project code style
* Implement `hashCode()` & `equals()` for `Person` & `Address` testing models
* Fix `ReactiveRedisStreamMessageProducerTests` not creating bean for `ConnectionFactory`.
Otherwise it caused to create one more not controlled `RedisClient`
* Code style clean up in the `ReactiveRedisStreamMessageProducer`
* Implement a group creation logic as a reactive stream deferring the call until
a subscription happens on the final `messageFlux`
* Move a common code for message building as the last `map()` operator in the final `Flux`
* Remove an `IntegrationFlow` definition in the `ReactiveRedisStreamMessageProducerTests`
as redundant
- Simplify MBeans - instead of wrapping to expose lifecycle methods,
implement `ManageableLifecycle`. Register an additional MBean for
polled endpoints to control the lifecycle.
* Polishing
- Move `QueueChannel` `@ManagedAttribute`s to `QueueChannelOperations`
- Make all `AbstractEndpoints` `IntegrationManagedResource`s and remove `ManagedEndpoint`
to allow exposure of any `@Managed*` methods (including those on `Pausable`)
- Revert to `Lifecycle` for classes that are not related to endpoints
- Remove legacy metrics from docs
The `ArrayList` cannot be used in the test-case for checking
its size because the property is not `volatile`: even if
content of the list is what we expect (test logs confirms that),
the non-`volatile` property doesn't give us an actual value
* Use `CopyOnWriteArrayList` for this concurrent test to
be sure that internal `array.length` reflects the reality
after each interaction with the list
In preparation for https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/303
Now that `CorrelationData` has a `Future<?>`, users might simply add
correlation data in a header and not receive confirm/return messages.
- No longer require channels for returns and confirms
- don't build the confirm message if there are no channels
- reduce the log level for no channels to DEBUG
- complete the user's future when a message is returned (async GW)
* Fix typo.
* Add gauges for queue channel size
The `QueueChannel` provides a current size and remaining capacity metrics
* Add Micrometer gauges into `QueueChannel` to expose the current values
of the size and remaining capacity
**Cherry-pick to 5.3.x, 5.2.x & 5.1.x**
* * Revert `@SuppressWarnings("unchecked")` for test
* Document new gauges for queue channel
* * Fix IntegrationManagementConfigurer for NPE on `metricsCaptor`
* Fix wording in meter descriptions
Co-authored-by: Michel Jung <michel.jung89@gmail.com>
Co-authored-by: Michel Jung <michel.jung89@gmail.com>
* Restore `Mono<RSocketRequester> getRSocketRequester()` on
`ClientRSocketConnector` and deprecate it in favor of newly introduced
`RSocketRequester getRequester()`: there is no need in wrapping a
`RSocketRequester` to deferred `Mono` since internal logic of the
`RSocketRequester` on client side is based on lazy-load API of the
`RSocketClient`
Since `RSocketRequester` is now lazy load on client side
there is no need to wrap it into a `Mono`.
* Change `getRSocketRequester()` to a plain getter
If there is a requirement to force connect to the server for receiving
requests from there, the `ClientRSocketConnector.connect()` should be used
Resolves https://github.com/spring-projects/spring-integration/issues/3340
- add common super-interface for MQTT components
- add `getConnectionInfo()` so users can examine server URIs etc
* Reinstate per-adapter URIs - support multiple
* Restore single URL per adapter.
* Code cleanup for previous commit.
* GH-3334: Add "embedded reaper" into CorrelationMH
Fixes https://github.com/spring-projects/spring-integration/issues/3334
* Add `expireTimeout` property into `AbstractCorrelatingMessageHandler`
to call newly introduced `purgeOrphanedGroups()` API for removing old
groups from the store
* Add `expireDuration` to perform `purgeOrphanedGroups()` task periodically
* * Add Java DSL and XML support for `expireTimeout` and `expireDuration` options
* Document the new feature
* * Fix language in docs
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* GH-3344: Treat kotlin.Unit return as null in MMIH
Fixes https://github.com/spring-projects/spring-integration/issues/3344
When function lambda doesn't return anything (e.g. a `void` method call is the last one),
Kotlin produces a `kotlin.Unit` instance as a return value which is not null and produced
as a reply message payload.
* Fix `MessagingMethodInvokerHelper` to treat a `kotlin.Unit` as `null` for reply
making Kotlin lambdas working the same way as Java lambdas when we don't return anything
from from there
**Cherry-pick to `5.3.x`**
* * Introduce `ClassUtils.isKotlinUnit(Class)` API;
use it in the `MessagingMethodInvokerHelper` instead of
`.getName().equals()`
* * Fix since on new `isKotlinUnit()` API
* Fix HTTP & WebFlux for a new CORS logic when `*` pattern
is used together with `allowCredentials`
* Fix RSocket module for deprecated API in SF
* The deprecated RMI API is left for the upcoming changes
Thanks to the pair of `CLIENT_ID`/`CREATED_DATE`,
it is possible to know if the lock is expired, already held or can be reacquire.
I think that the pre delete row is unnecessary and redundant with these two columns.
The rows doesn't need to be purged because the number of row in the table is finite and equal to the number of locks.
At worst, the table will be able to be purged by `deleteExpired()`.
I propose to overwrite these two columns at acquire lock time instead of pre delete the row.
So `delete` then `update` or `insert` become `update` or `insert`.
One client held a lock.
- Another client will be able to hold the same lock only if:
1. the row doesn't exist. Done when the first client will call `unlock()`.
2. the lock expire in case of hardware shutdown. Done by "`OR CREATED_DATE<?`" in `updateQuery`.
- The same client will be able to reacquire it only if:
1. the row doesn't exist.
2. the lock is already held by him. Done by "`CLIENT_ID=? OR`" in `updateQuery`.
**Cost with PostgreSQL:**
```
EXPLAIN DELETE FROM INT_LOCK WHERE REGION='DEFAULT' AND LOCK_KEY='FOO' AND CREATED_DATE<'2020-07-14';
Delete on int_lock (cost=0.14..8.17 rows=1 width=6)
-> Index Scan using int_lock_pk on int_lock (cost=0.14..8.17 rows=1 width=6)
Index Cond: (((region)::text = 'DEFAULT'::text) AND (lock_key = 'FOO'::bpchar))
Filter: (created_date < '2020-07-14 00:00:00'::timestamp without time zone)
EXPLAIN UPDATE INT_LOCK SET CREATED_DATE='2020-07-15' WHERE REGION='DEFAULT' AND LOCK_KEY='FOO' AND CLIENT_ID=NULL;
Update on int_lock (cost=0.00..11.40 rows=1 width=520)
-> Result (cost=0.00..11.40 rows=1 width=520)
One-Time Filter: false
-> Seq Scan on int_lock (cost=0.00..11.40 rows=1 width=520)
```
```
EXPLAIN UPDATE INT_LOCK SET CLIENT_ID=NULL, CREATED_DATE='2020-07-15' WHERE REGION='DEFAULT' AND LOCK_KEY='FOO' AND (CLIENT_ID=NULL OR CREATED_DATE<'2020-07-14')
Update on int_lock (cost=0.14..8.17 rows=1 width=372)
-> Index Scan using int_lock_pk on int_lock (cost=0.14..8.17 rows=1 width=372)
Index Cond: (((region)::text = 'DEFAULT'::text) AND (lock_key = 'FOO'::bpchar))
Filter: (created_date < '2020-07-14 00:00:00'::timestamp without time zone)
```
