Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2880: Handle Pausable in Control Bus (#2940)

Browse Source 
* GH-2880: Handle `Pausable` in Control Bus

Fixes https://github.com/spring-projects/spring-integration/issues/2880

* Refactor `ControlBusMethodFilter` to handle `Pausable` managed operations
* Optimize and internal `ControlBusMethodFilter.filter()` logic to rely
on the `MergedAnnotations`
* Modify `EnableIntegrationTests` to test new functionality and document
the feature

* * `ControlBusMethodFilter` to deal with plain `Lifecycle` impls as well
This commit is contained in:
Artem Bilan
2019-05-30 14:10:01 -04:00
committed by Gary Russell
parent e9591c6fdf
commit 374b4b70f1
4 changed files with 54 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
spring-integration-core/src/main/java/org/springframework/integration/expression/ControlBusMethodFilter.java
View File

@@ -16,14 +16,16 @@
package org.springframework.integration.expression;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import org.springframework.context.Lifecycle;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.AnnotationFilter;
import org.springframework.core.annotation.MergedAnnotations;
import org.springframework.core.annotation.RepeatableContainers;
import org.springframework.expression.MethodFilter;
import org.springframework.integration.endpoint.Pausable;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.util.CustomizableThreadCreator;
@@ -32,7 +34,7 @@ import org.springframework.util.ReflectionUtils;
/**
* SpEL {@link MethodFilter} to restrict method invocations to:
* <ul>
* <li> {@link Lifecycle} components
* <li> {@link Pausable} or {@link Lifecycle} components
* <li> {@code get}, {@code set} and {@code shutdown} methods of {@link CustomizableThreadCreator}
* <li> methods with {@link ManagedAttribute} and {@link ManagedOperation} annotations
* </ul>
@@ -40,14 +42,15 @@ import org.springframework.util.ReflectionUtils;
*
* @author Mark Fisher
* @author Artem Bilan
* @since 4.0
*/
*
* @since 4.0
*/
public class ControlBusMethodFilter implements MethodFilter {
public List<Method> filter(List<Method> methods) {
List<Method> supportedMethods = new ArrayList<Method>();
List<Method> supportedMethods = new ArrayList<>();
for (Method method : methods) {
if (this.accept(method)) {
if (accept(method)) {
supportedMethods.add(method);
}
}
@@ -56,23 +59,25 @@ public class ControlBusMethodFilter implements MethodFilter {
private boolean accept(Method method) {
Class<?> declaringClass = method.getDeclaringClass();
if (Lifecycle.class.isAssignableFrom(declaringClass)
&& ReflectionUtils.findMethod(Lifecycle.class, method.getName(), method.getParameterTypes()) != null) {
String methodName = method.getName();
if ((Pausable.class.isAssignableFrom(declaringClass) || Lifecycle.class.isAssignableFrom(declaringClass))
&& ReflectionUtils.findMethod(Pausable.class, methodName, method.getParameterTypes()) != null) {
return true;
}
if (CustomizableThreadCreator.class.isAssignableFrom(declaringClass)
&& (method.getName().startsWith("get")
|| method.getName().startsWith("set")
|| method.getName().startsWith("shutdown"))) {
&& (methodName.startsWith("get")
|| methodName.startsWith("set")
|| methodName.startsWith("shutdown"))) {
return true;
}
if (this.hasAnnotation(method, ManagedAttribute.class) || this.hasAnnotation(method, ManagedOperation.class)) {
return true;
}
return false;
MergedAnnotations mergedAnnotations =
MergedAnnotations.from(method, MergedAnnotations.SearchStrategy.EXHAUSTIVE,
RepeatableContainers.none(), AnnotationFilter.PLAIN);
return mergedAnnotations.get(ManagedAttribute.class).isPresent()
|| mergedAnnotations.get(ManagedOperation.class).isPresent();
}
private boolean hasAnnotation(Method method, Class<? extends Annotation> annotationType) {
return AnnotationUtils.findAnnotation(method, annotationType) != null;
}
}

34
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -94,6 +94,7 @@ import org.springframework.integration.core.MessageSource;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.endpoint.Pausable;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.expression.SpelPropertyAccessorRegistrar;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
@@ -409,11 +410,16 @@ public class EnableIntegrationTests {
assertThat(message.getHeaders().get("foo")).isEqualTo("FOO");
MessagingTemplate messagingTemplate = new MessagingTemplate(this.controlBusChannel);
assertThat(messagingTemplate.convertSendAndReceive("@lifecycle.isRunning()", Boolean.class)).isEqualTo(false);
this.controlBusChannel.send(new GenericMessage<>("@lifecycle.start()"));
assertThat(messagingTemplate.convertSendAndReceive("@lifecycle.isRunning()", Boolean.class)).isEqualTo(true);
this.controlBusChannel.send(new GenericMessage<>("@lifecycle.stop()"));
assertThat(messagingTemplate.convertSendAndReceive("@lifecycle.isRunning()", Boolean.class)).isEqualTo(false);
assertThat(messagingTemplate.convertSendAndReceive("@pausable.isRunning()", Boolean.class)).isEqualTo(false);
this.controlBusChannel.send(new GenericMessage<>("@pausable.start()"));
assertThat(messagingTemplate.convertSendAndReceive("@pausable.isRunning()", Boolean.class)).isEqualTo(true);
this.controlBusChannel.send(new GenericMessage<>("@pausable.stop()"));
assertThat(messagingTemplate.convertSendAndReceive("@pausable.isRunning()", Boolean.class)).isEqualTo(false);
this.controlBusChannel.send(new GenericMessage<>("@pausable.pause()"));
Object pausable = this.context.getBean("pausable");
assertThat(TestUtils.getPropertyValue(pausable, "paused", Boolean.class)).isTrue();
this.controlBusChannel.send(new GenericMessage<>("@pausable.resume()"));
assertThat(TestUtils.getPropertyValue(pausable, "paused", Boolean.class)).isFalse();
Map<String, ServiceActivatingHandler> beansOfType =
this.context.getBeansOfType(ServiceActivatingHandler.class);
@@ -638,7 +644,7 @@ public class EnableIntegrationTests {
@Test
public void testMonoGateway() throws Exception {
final AtomicReference<List<Integer>> ref = new AtomicReference<List<Integer>>();
final AtomicReference<List<Integer>> ref = new AtomicReference<>();
final CountDownLatch consumeLatch = new CountDownLatch(1);
Flux.just("1", "2", "3", "4", "5")
@@ -1035,11 +1041,13 @@ public class EnableIntegrationTests {
}
@Bean
public Lifecycle lifecycle() {
return new Lifecycle() {
public Pausable pausable() {
return new Pausable() {
private volatile boolean running;
private volatile boolean paused;
@Override
public void start() {
this.running = true;
@@ -1055,6 +1063,16 @@ public class EnableIntegrationTests {
return this.running;
}
@Override
public void pause() {
this.paused = true;
}
@Override
public void resume() {
this.paused = false;
}
};
}