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

Add controlBus() EIP-method

Browse Source
This commit is contained in:
Artem Bilan
2014-03-12 22:32:42 +02:00
parent e738864125
commit 53b25dbdcf
2 changed files with 64 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
View File

@@ -16,9 +16,16 @@
package org.springframework.integration.dsl;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.Lifecycle;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.expression.MethodFilter;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
@@ -44,6 +51,7 @@ import org.springframework.integration.filter.MethodInvokingSelector;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.handler.DelayHandler;
import org.springframework.integration.handler.ExpressionCommandMessageProcessor;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.splitter.AbstractMessageSplitter;
import org.springframework.integration.splitter.DefaultMessageSplitter;
@@ -57,9 +65,13 @@ import org.springframework.integration.transformer.HeaderFilter;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.MethodInvokingTransformer;
import org.springframework.integration.transformer.Transformer;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.CustomizableThreadCreator;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
@@ -114,6 +126,10 @@ public final class IntegrationFlowBuilder {
return this.channel(messageChannelSpec.get());
}
public IntegrationFlowBuilder controlBus() {
return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(new ControlBusMethodFilter())), null);
}
public IntegrationFlowBuilder transform(String expression) {
Assert.hasText(expression);
return this.transform(new ExpressionEvaluatingTransformer(PARSER.parseExpression(expression)));
@@ -438,4 +454,40 @@ public final class IntegrationFlowBuilder {
return this.flow;
}
private static class ControlBusMethodFilter implements MethodFilter {
public List<Method> filter(List<Method> methods) {
List<Method> supportedMethods = new ArrayList<Method>();
for (Method method : methods) {
if (this.accept(method)) {
supportedMethods.add(method);
}
}
return supportedMethods;
}
private boolean accept(Method method) {
Class<?> declaringClass = method.getDeclaringClass();
if (Lifecycle.class.isAssignableFrom(declaringClass)
&& ReflectionUtils.findMethod(Lifecycle.class, method.getName(), method.getParameterTypes()) != null) {
return true;
}
if (CustomizableThreadCreator.class.isAssignableFrom(declaringClass)
&& (method.getName().startsWith("get")
|| method.getName().startsWith("set")
|| method.getName().startsWith("shutdown"))) {
return true;
}
if (this.hasAnnotation(method, ManagedAttribute.class) || this.hasAnnotation(method, ManagedOperation.class)) {
return true;
}
return false;
}
private boolean hasAnnotation(Method method, Class<? extends Annotation> annotationType) {
return AnnotationUtils.findAnnotation(method, annotationType) != null;
}
}
}

16
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
View File

@@ -50,7 +50,6 @@ import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
@@ -110,6 +109,10 @@ public class IntegrationFlowTests {
@Autowired
private ListableBeanFactory beanFactory;
@Autowired
@Qualifier("controlBus")
private MessageChannel controlBus;
@Autowired
@Qualifier("flow1QueueChannel")
private PollableChannel outputChannel;
@@ -207,7 +210,7 @@ public class IntegrationFlowTests {
assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
}
this.beanFactory.getBean("payloadSerializingTransformer", Lifecycle.class).start();
this.controlBus.send(new GenericMessage<Object>("@payloadSerializingTransformer.start()"));
final AtomicBoolean used = new AtomicBoolean();
@@ -253,7 +256,7 @@ public class IntegrationFlowTests {
assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
}
this.beanFactory.getBean("bridge", Lifecycle.class).start();
this.controlBus.send(new GenericMessage<Object>("@bridge.start()"));
this.bridgeFlow2Input.send(message);
reply = this.bridgeFlow2Output.receive(5000);
assertNotNull(reply);
@@ -396,7 +399,7 @@ public class IntegrationFlowTests {
assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
}
this.beanFactory.getBean("xpathHeaderEnricher", Lifecycle.class).start();
this.controlBus.send(new GenericMessage<Object>("@xpathHeaderEnricher.start()"));
this.xpathHeaderEnricherInput.send(message);
Message<?> result = replyChannel.receive(2000);
@@ -418,6 +421,11 @@ public class IntegrationFlowTests {
return source;
}
@Bean
public IntegrationFlow controlBusFlow() {
return IntegrationFlows.from("controlBus").controlBus().get();
}
@Bean
public IntegrationFlow flow1() {
return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100)))