GH-9381: Introduce Control Bus commands management

Fixes: #9381

Currently, there is no way to know in one place what Control Bus commands are available and with what arguments

* Add `ControlBusCommandRegistry` infrastructure bean to gather control bus commands from beans and expose them for invocation
* Add `ControlBusController` to expose a `/control-bus` REST service against the mentioned `ControlBusCommandRegistry`
* Add `@EnableIntegrationManagement(loadControlBusCommands)` to be able to load all the Control Bus commands from the application context instead of on demand by default
* Deprecated existing SpEL(and Groovy)-based Control Bus functionality in favor of new, more manageable, logic
This commit is contained in:
Artem Bilan
2024-08-08 11:07:56 -04:00
parent 77e3b08d16
commit 4d787554b8
81 changed files with 1826 additions and 855 deletions

View File

@@ -159,6 +159,7 @@
** xref:http/proxy.adoc[]
** xref:http/header-mapping.adoc[]
** xref:http/int-graph-controller.adoc[]
** xref:http/control-bus-controller.adoc[]
** xref:http/samples.adoc[]
* xref:ip.adoc[]
** xref:ip/intro.adoc[]

View File

@@ -71,7 +71,7 @@ See xref:claim-check.adoc[Claim Check] for more details.
We have provided implementations of the https://www.enterpriseintegrationpatterns.com/ControlBus.html[control bus] pattern, which lets you use messaging to manage and monitor endpoints and channels.
The implementations include both a SpEL-based approach and one that runs Groovy scripts.
See xref:groovy.adoc#groovy-control-bus[Control Bus] and xref:groovy.adoc#groovy-control-bus[Control Bus] for more details.
See xref:control-bus.adoc[Control Bus] for more details.
[[new-adapters]]
== New Channel Adapters and Gateways

View File

@@ -66,7 +66,7 @@ The `splitter` now supports a `discardChannel` configuration option.
See xref:splitter.adoc[Splitter] for more information.
The Control Bus can now handle `Pausable` (extension of `Lifecycle`) operations.
See xref:groovy.adoc#groovy-control-bus[Control Bus] for more information.
See xref:control-bus.adoc[Control Bus] for more information.
The `Function<MessageGroup, Map<String, Object>>` strategy has been introduced for the aggregator component to merge and compute headers for output messages.
See xref:aggregator.adoc#aggregator-api[Aggregator Programming Model] for more information.

View File

@@ -120,7 +120,7 @@ The `MessageHandler` instances (`MessageSource` instances) are also eligible to
Starting with version 4.0, all messaging annotations provide `SmartLifecycle` options (`autoStartup` and `phase`) to allow endpoint lifecycle control on application context initialization.
They default to `true` and `0`, respectively.
To change the state of an endpoint (such as `start()` or `stop()`), you can obtain a reference to the endpoint bean by using the `BeanFactory` (or autowiring) and invoke the methods.
Alternatively, you can send a command message to the `Control Bus` (see xref:groovy.adoc#groovy-control-bus[Control Bus]).
Alternatively, you can send a command message to the xref:control-bus.adoc[Control Bus].
For these purposes, you should use the `beanName` mentioned earlier in the preceding paragraph.
[IMPORTANT]

View File

@@ -42,7 +42,7 @@ By default, only `MessageHeaders.ID` and `MessageHeaders.TIMESTAMP` are not copi
Since version 4.3.2.
<7> A comma-separated list of `AbstractEndpoint` bean names patterns (`xxx*`, `*xxx`, `*xxx*` or `xxx*yyy`) that should not be started automatically during application startup.
You can manually start these endpoints later by their bean name through a `Control Bus` (see xref:groovy.adoc#groovy-control-bus[Control Bus]), by their role with the `SmartLifecycleRoleController` (see xref:endpoint.adoc#endpoint-roles[Endpoint Roles]), or by `Lifecycle` bean injection.
You can manually start these endpoints later by their bean name through a xref:control-bus.adoc[Control Bus], by their role with the `SmartLifecycleRoleController` (see xref:endpoint-roles.adoc[Endpoint Roles]), or by `Lifecycle` bean injection.
You can explicitly override the effect of this global property by specifying `auto-startup` XML annotation or the `autoStartup` annotation attribute or by calling `AbstractEndpoint.setAutoStartup()` in the bean definition.
Since version 4.3.12.

View File

@@ -199,7 +199,7 @@ These methods can be invoked directly by getting a reference to the registry, or
[source]
----
"@integrationHeaderChannelRegistry.runReaper()"
"integrationHeaderChannelRegistry.runReaper"
----
This sub-element is a convenience, and is the equivalent of specifying the following configuration:

View File

@@ -4,6 +4,8 @@
As described in the https://www.enterpriseintegrationpatterns.com/[_Enterprise Integration Patterns_] (EIP) book, the idea behind the control bus is that the same messaging system can be used for monitoring and managing the components within the framework as is used for "`application-level`" messaging.
In Spring Integration, we build upon the adapters described above so that you can send messages as a means of invoking exposed operations.
IMPORTANT: Since Control Bus is powerful enough to make changes into the system state, it is recommended to secure its messages reception (see `SecurityContextChannelInterceptor`) and expose Control Bus management (message source) only into DMZ.
The following example shows how to configure a control bus with XML:
[source,xml]
@@ -15,33 +17,84 @@ The control bus has an input channel that can be accessed for invoking operation
It also has all the common properties of a service activating endpoint.
For example, you can specify an output channel if the result of the operation has a return value that you want to send on to a downstream channel.
The control bus runs messages on the input channel as Spring Expression Language (SpEL) expressions.
It takes a message, compiles the body to an expression, adds some context, and then runs it.
The default context supports any method that has been annotated with `@ManagedAttribute` or `@ManagedOperation`.
It also supports the methods on Spring's `Lifecycle` interface (and its `Pausable` extension since version 5.2), and it supports methods that are used to configure several of Spring's `TaskExecutor` and `TaskScheduler` implementations.
The control bus runs messages on the input channel as a managed operation in a simple string format like `beanName.methodName`.
The arguments for the target method parameters must be supplied as a list in the `IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS` header.
The bean and the method to call is resolved from the `ControlBusCommandRegistry` infrastructure bean.
By default, the `ControlBusCommandRegistry` registers commands on demand: its `eagerInitialization` flag can be turned on via `@EnableIntegrationManagement(loadControlBusCommands = "true")`.
The functionality of Control Bus is similar to JMX, therefore method eligibility for command must honor these requirements:
- The method that has been annotated with `@ManagedAttribute` or `@ManagedOperation`;
- Spring's `Lifecycle` interface (and its `Pausable` extension since version 5.2);
- The methods that are used to configure several of Spring's `TaskExecutor` and `TaskScheduler` implementations.
The simplest way to ensure that your own methods are available to the control bus is to use the `@ManagedAttribute` or `@ManagedOperation` annotations.
Since those annotations are also used for exposing methods to a JMX MBean registry, they offer a convenient by-product: Often, the same types of operations you want to expose to the control bus are reasonable for exposing through JMX).
Resolution of any particular instance within the application context is achieved in the typical SpEL syntax.
To do so, provide the bean name with the SpEL prefix for beans (`@`).
For example, to execute a method on a Spring Bean, a client could send a message to the operation channel as follows:
Since those annotations are also used for exposing methods to a JMX MBean registry, they offer a convenient by-product: often, the same types of operations you want to expose to the control bus are reasonable for exposing through JMX).
See more information in the `ControlBusCommandRegistry` and `ControlBusMethodFilter` Javadocs.
To execute a method on a Spring Bean, a client could send a message to the operation channel as follows:
[source,java]
----
Message operation = MessageBuilder.withPayload("@myServiceBean.shutdown()").build();
operationChannel.send(operation)
Message<?> operation = MessageBuilder.withPayload("myServiceBean.shutdown").build();
operationChannel.send(operation);
----
The root of the context for the expression is the `Message` itself, so you also have access to the `payload` and `headers` as variables within your expression.
This is consistent with all the other expression support in Spring Integration endpoints.
If target method to call has arguments (e.g. `ThreadPoolTaskExecutor.setMaxPoolSize(int maxPoolSize)`), those values has to be provided as `IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS` header:
With Java annotations, you can configured the control bus as follows:
[source,java]
----
Message<?> operation =
MessageBuilder.withPayload("myTaskExecutor.setMaxPoolSize")
.setHeader(IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS, List.of(10))
.build();
operationChannel.send(operation);
----
You can think about these commands as `PreparedStatement` instances in JDBC with parameter binding.
The types of arguments must match types of method parameters.
They are used as additional criteria to select a method to call according to Java method overloading feature.
For example the component:
[source,java]
----
@ManagedResource
class TestManagementComponent {
@ManagedOperation
public void operation() {
}
@ManagedOperation(description = "The overloaded operation with int argument")
public void operation(int input) {
}
@ManagedOperation(description = "The overloaded operation with two arguments")
public void operation(int input1, String input2) {
}
@ManagedOperation
public int operation2() {
return 123;
}
}
----
will expose 3 commands with `operation` name.
When we call `testManagementComponent.operation` command, we should choose a proper list of values for the `IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS` header to let the `ControlBusCommandRegistry` to filter out the target method on the bean.
With Java annotations, you can configure the control bus as follows:
[source,java]
----
@Bean
@ServiceActivator(inputChannel = "operationChannel")
public ExpressionControlBusFactoryBean controlBus() {
return new ExpressionControlBusFactoryBean();
public ControlBusFactoryBean controlBus() {
return new ControlBusFactoryBean();
}
----
@@ -68,3 +121,5 @@ public IntegrationFlow controlBus() {
----
In this case, the channel is named `controlBus.input`.
Also, see xref:http/control-bus-controller.adoc[Control Bus REST Controller] for exposing Control Bus management over HTTP.

View File

@@ -192,7 +192,7 @@ These operations can be invoked through a `Control Bus` command, as the followin
[source,java]
----
Message<String> delayerReschedulingMessage =
MessageBuilder.withPayload("@'delayer.handler'.reschedulePersistedMessages()").build();
MessageBuilder.withPayload("'delayer.handler'.reschedulePersistedMessages").build();
controlBusChannel.send(delayerReschedulingMessage);
----

View File

@@ -129,7 +129,7 @@ This means that the poller continues calling `receive()` without waiting, until
For example, if a poller has a ten-second interval trigger and a `maxMessagesPerPoll` setting of `25`, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds.
It grabs 25, waits ten seconds, grabs the next 25, and so on.
If `maxMessagesPerPoll` is configured with a negative value, then `MessageSource.receive()` is called within a single polling cycle until it returns `null`.
Starting with version 5.5, a `0` value has a special meaning - skip the `MessageSource.receive()` call altogether, which may be considered as pausing for this polling endpoint until the `maxMessagesPerPoll` is changed to a n non-zero value at a later time, e.g. via a Control Bus.
Starting with version 5.5, a `0` value has a special meaning - skip the `MessageSource.receive()` call altogether, which may be considered as pausing for this polling endpoint until the `maxMessagesPerPoll` is changed to a non-zero value at a later time, e.g. via a Control Bus.
The `receiveTimeout` property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation.
For example, consider two options that seem similar on the surface but are actually quite different: The first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds, while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds.

View File

@@ -18,7 +18,7 @@ When all files are consumed, the remote fetch is attempted again, to pick up any
IMPORTANT: When you deploy multiple instances of an application, we recommend a small `max-fetch-size`, to avoid one instance "`grabbing`" all the files and starving other instances.
Another use for `max-fetch-size` is if you want to stop fetching remote files but continue to process files that have already been fetched.
Setting the `maxFetchSize` property on the `MessageSource` (programmatically, with JMX, or with a xref:groovy.adoc#groovy-control-bus[control bus]) effectively stops the adapter from fetching more files but lets the poller continue to emit messages for files that have previously been fetched.
Setting the `maxFetchSize` property on the `MessageSource` (programmatically, with JMX, or with a xref:control-bus.adoc[control bus]) effectively stops the adapter from fetching more files but lets the poller continue to emit messages for files that have previously been fetched.
If the poller is active when the property is changed, the change takes effect on the next poll.
Starting with version 5.1, the synchronizer can be provided with a `Comparator<FTPFile>`.

View File

@@ -139,37 +139,3 @@ NOTE: Using `compilerConfiguration` does not automatically add an `ASTTransforma
If you still need `CompileStatic`, you should manually add a `new ASTTransformationCustomizer(CompileStatic.class)` into the `CompilationCustomizers` of that custom `compilerConfiguration`.
NOTE: The Groovy compiler customization does not have any effect on the `refresh-check-delay` option, and reloadable scripts can be statically compiled, too.
[[groovy-control-bus]]
== Control Bus
As described in (https://www.enterpriseintegrationpatterns.com/ControlBus.html[Enterprise Integration Patterns]), the idea behind the control bus is that you can use the same messaging system for monitoring and managing the components within the framework as is used for "`application-level`" messaging.
In Spring Integration, we build upon the adapters described earlier so that you can send Messages as a means of invoking exposed operations.
One option for those operations is Groovy scripts.
The following example configures a Groovy script for the control bus:
[source,xml]
----
<int-groovy:control-bus input-channel="operationChannel"/>
----
The control bus has an input channel that can be accessed to invoke operations on the beans in the application context.
The Groovy control bus runs messages on the input channel as Groovy scripts.
It takes a message, compiles the body to a script, customizes it with a `GroovyObjectCustomizer`, and runs it.
The control bus' `MessageProcessor` exposes all beans in the application context that are annotated with `@ManagedResource` and implement Spring's `Lifecycle` interface or extend Spring's `CustomizableThreadCreator` base class (for example, several of the `TaskExecutor` and `TaskScheduler` implementations).
IMPORTANT: Be careful about using managed beans with custom scopes (such as 'request') in the Control Bus' command scripts, especially inside an asynchronous message flow.
If `MessageProcessor` of the control bus cannot expose a bean from the application context, you may end up with some `BeansException` during the command script's run.
For example, if a custom scope's context is not established, the attempt to get a bean within that scope triggers a `BeanCreationException`.
If you need to further customize the Groovy objects, you can also provide a reference to a bean that implements `GroovyObjectCustomizer` through the `customizer` attribute, as the following example shows:
[source,xml]
----
<int-groovy:control-bus input-channel="input"
output-channel="output"
customizer="groovyCustomizer"/>
<beans:bean id="groovyCustomizer" class="org.foo.MyGroovyObjectCustomizer"/>
----

View File

@@ -0,0 +1,119 @@
[[control-bus-controller]]
= Control Bus Controller
:page-section-summary-toc: 1
Starting with version 6.4, the HTTP module provides an `@EnableControlBusController` configuration class annotation to expose the `ControlBusController` as a REST service at the `/control-bus` path.
The `ControlBusControllerConfiguration` underneath enables eager initialization for the `ControlBusCommandRegistry` to expose all the available control bus commands for the mentioned REST service.
The `/control-bus` GET request returns all the control bus commands for the application in a format like this:
[source,json]
----
[
{
"beanName": "errorChannel",
"commands": [
{
"command": "errorChannel.setShouldTrack",
"description": "setShouldTrack",
"parameterTypes": [
"boolean"
]
},
{
"command": "errorChannel.setLoggingEnabled",
"description": "Use to disable debug logging during normal message flow",
"parameterTypes": [
"boolean"
]
},
{
"command": "errorChannel.isLoggingEnabled",
"description": "isLoggingEnabled",
"parameterTypes": []
}
]
},
{
"beanName": "testManagementComponent",
"commands": [
{
"command": "testManagementComponent.operation2",
"description": "operation2",
"parameterTypes": []
},
{
"command": "testManagementComponent.operation",
"description": "operation",
"parameterTypes": []
},
{
"command": "testManagementComponent.operation",
"description": "operation",
"parameterTypes": [
"int",
"java.lang.String"
]
},
{
"command": "testManagementComponent.operation",
"description": "operation",
"parameterTypes": [
"int"
]
}
]
}
]
----
Essentially, a JSON-serialized list of `ControlBusController.ControlBusBean` instances.
Each entry is a bean with a list of control bus eligible methods (see `ControlBusMethodFilter` for more information) with their parameter types and description from the `@ManagedOperation` or `@ManagedAttribute` (falls back to method name otherwise).
The GET method of this REST controller for `/control-bus/{beanName}` returns commands for specific bean.
The POST method to `/control-bus/{beanName.methodName}` invokes the command.
The body of the request may contain a list of values and their types for command to execute.
For example, the `operation` command with `int` argument for the class:
[source,java]
----
@ManagedResource
class TestManagementComponent {
@ManagedOperation
public void operation() {
}
@ManagedOperation(description = "The overloaded operation with int argument")
public void operation(int input) {
}
@ManagedOperation(description = "The overloaded operation with two arguments")
public void operation(int input1, String input2) {
}
@ManagedOperation
public int operation2() {
return 123;
}
}
----
could be called like `/testManagementComponent.operation` using mention POST method with body:
[source,json]
----
[
{
"value": "1",
"parameterType": "int"
}
]
----
See xref:control-bus.adoc[Control Bus] for more information.

View File

@@ -97,7 +97,7 @@ In fact, it applies to every other router, including expression-based routers, s
Any router that is a subclass of the `AbstractMappingMessageRouter` (which includes most framework-defined routers) is a dynamic router, because the `channelMapping` is defined at the `AbstractMappingMessageRouter` level.
That map's setter method is exposed as a public method along with the 'setChannelMapping' and 'removeChannelMapping' methods.
These let you change, add, and remove router mappings at runtime, as long as you have a reference to the router itself.
It also means that you could expose these same configuration options through JMX (see xref:jmx.adoc[JMX Support]) or the Spring Integration control bus (see xref:groovy.adoc#groovy-control-bus[Control Bus]) functionality.
It also means that you could expose these same configuration options through JMX (see xref:jmx.adoc[JMX Support]) or the Spring Integration control bus (see xref:control-bus.adoc[Control Bus]) functionality.
IMPORTANT: Falling back to the channel key as the channel name is flexible and convenient.
However, if you don't trust the message creator, a malicious actor (who has knowledge of the system) could create a message that is routed to an unexpected channel.
@@ -109,7 +109,7 @@ You may therefore wish to disable this feature (set the `channelKeyFallback` pro
One way to manage the router mappings is through the https://www.enterpriseintegrationpatterns.com/ControlBus.html[control bus] pattern, which exposes a control channel to which you can send control messages to manage and monitor Spring Integration components, including routers.
NOTE: For more information about the control bus, see xref:groovy.adoc#groovy-control-bus[Control Bus].
NOTE: For more information about the control bus, see xref:control-bus.adoc[Control Bus].
Typically, you would send a control message asking to invoke a particular operation on a particular managed component (such as a router).
The following managed operations (methods) are specific to changing the router resolution process:
@@ -125,15 +125,19 @@ The following methods let you do so:
* `public Map<String, String>getChannelMappings()`: Returns the current mappings.
* `public void replaceChannelMappings(Properties channelMappings)`: Updates the mappings.
Note that the `channelMappings` parameter is a `Properties` object.
This arrangement lets a control bus command use the built-in `StringToPropertiesConverter`, as the following example shows:
Note that the `channelMappings` parameter is a `Properties` object, so this has to be added to the respective `IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS` header:
[source]
----
"@'router.handler'.replaceChannelMappings('foo=qux \n baz=bar')"
Properties newMapping = new Properties();
newMapping.setProperty("foo", "bar");
newMapping.setProperty("baz", "qux");
Message<?> replaceChannelMappingsCommandMessage =
MessageBuilder.withPayload("'router.handler'.replaceChannelMappings")
.setHeader(IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS, List.of(newMapping))
.build();
----
Note that each mapping is separated by a newline character (`\n`).
For programmatic changes to the map, we recommend that you use the `setChannelMappings` method, due to type-safety concerns.
`replaceChannelMappings` ignores keys or values that are not `String` objects.

View File

@@ -279,7 +279,7 @@ If this attribute is not defined, the channel is always among the list of recipi
Starting with version 4.1, the `RecipientListRouter` provides several operations to manipulate recipients dynamically at runtime.
These management operations are presented by `RecipientListRouterManagement` through the `@ManagedResource` annotation.
They are available by using xref:groovy.adoc#groovy-control-bus[Control Bus] as well as by using JMX, as the following example shows:
They are available by using xref:control-bus.adoc[Control Bus] as well as by using JMX, as the following example shows:
[source,xml]
----
@@ -293,7 +293,10 @@ They are available by using xref:groovy.adoc#groovy-control-bus[Control Bus] as
----
[source,java]
----
messagingTemplate.convertAndSend(controlBus, "@'simpleRouter.handler'.addRecipient('channel2')");
Message<?> addRecipientCommandMessage =
MessageBuilder.withPayload("'simpleRouter.handler'.addRecipient")
.setHeader(IntegrationMessageHeaderAccessor.CONTROL_BUS_ARGUMENTS, List.of("channel2"))
.build();
----
From the application start up the `simpleRouter`, has only one `channel1` recipient.

View File

@@ -315,7 +315,7 @@ When all files are consumed, the remote fetch is attempted again, to pick up any
IMPORTANT: When you deploy multiple instances of an application, we recommend a small `max-fetch-size`, to avoid one instance "`grabbing`" all the files and starving other instances.
Another use for `max-fetch-size` is if you want to stop fetching remote files but continue to process files that have already been fetched.
Setting the `maxFetchSize` property on the `MessageSource` (programmatically, with JMX, or with a xref:groovy.adoc#groovy-control-bus[control bus]) effectively stops the adapter from fetching more files but lets the poller continue to emit messages for files that have previously been fetched.
Setting the `maxFetchSize` property on the `MessageSource` (programmatically, with JMX, or with a xref:control-bus.adoc[control bus]) effectively stops the adapter from fetching more files but lets the poller continue to emit messages for files that have previously been fetched.
If the poller is active when the property is changed, the change takes effect on the next poll.
The synchronizer can be provided with a `Comparator<SmbFile>`.

View File

@@ -16,6 +16,13 @@ In general the project has been moved to the latest dependency versions.
[[x6.4-new-components]]
=== New Components
The new Control Bus interaction model is implemented in the `ControlBusCommandRegistry`.
A new `ControlBusFactoryBean` class is recommended to be used instead of deprecated `ExpressionControlBusFactoryBean`.
See xref:control-bus.adoc[Control Bus] for more information.
Also, a `ControlBusController` (together with an `@EnableControlBusController`) is introduced for managing exposed commands by the mentioned `ControlBusCommandRegistry`.
See xref:http.adoc[HTTP Support] for more information.
[[x6.4-general]]
=== General Changes
@@ -34,13 +41,22 @@ The byte array handling for serialized message is fully deferred to JDBC driver.
The `LockRepository.delete()` method return the result of removing ownership of a distributed lock.
And the `JdbcLockRegistry.JdbcLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired.
See xref:jdbc.adoc[JDBC Support] for more information.
[[x6.4-zeromq-changes]]
=== ZeroMQ Changes
The outbound component `ZeroMqMessageHandler` (and respective API) can now bind a TCP port instead of connecting to a given URL.
See xref:zeromq.adoc[ZeroMQ Support] for more information.
[[x6.4-redis-changes]]
=== Redis Changes
Instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired.
Instead of throwing `IllegalStateException`, the `RedisLockRegistry.RedisLock.unlock()` method throws `ConcurrentModificationException` if the ownership of the lock is expired.
See xref:redis.adoc[Redis Support] for more information.
[[x6.4-groovy-changes]]
=== Groovy Changes
The `ControlBusFactoryBean` (and respective `<int-groovy:control-bus>` XML tag) has been deprecated (for removal) in favor of new introduced `ControlBusFactoryBean` based on a new model implemented in the `ControlBusCommandRegistry`.
See xref:control-bus.adoc[Control Bus] for more information.