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

Introduce method-invoking handle

Browse Source 
* Add generic argument to `GenericEndpointSpec`.
Since `EndpointSpec#get()` returns `Tuple` with `EndpointFactoryBean` and `MessageHandler` objects
it is useful to get deal with specific generic for further `MessageHandler` configuration within `EndpointConfigurer` lambda
* Change DSL-methods to get deal deal with those generics.
* Add `FileWritingMessageHandler` test to demonstrate it.
* Fix bug around double `MessageHandler` bean registration from `DslIntegrationConfigurationInitializer`, when handler is a reference
to existing bean.
This commit is contained in:
Artem Bilan
2014-02-17 17:11:41 +02:00
parent ac46625a83
commit c161210c3e
13 changed files with 163 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java
View File

@@ -22,7 +22,7 @@ import org.springframework.messaging.MessageChannel;
/**
* @author Artem Bilan
* @since 4.0
*/
public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpointSpec, MessageFilter> {

8
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java
View File

@@ -21,11 +21,11 @@ import org.springframework.messaging.MessageHandler;
/**
* @author Artem Bilan
* @since 4.0
*/
public final class GenericEndpointSpec extends ConsumerEndpointSpec<GenericEndpointSpec, MessageHandler> {
GenericEndpointSpec(MessageHandler messageHandler) {
*/
public final class GenericEndpointSpec<H extends MessageHandler> extends ConsumerEndpointSpec<GenericEndpointSpec<H>, H> {
GenericEndpointSpec(H messageHandler) {
super(messageHandler);
}

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

@@ -25,6 +25,7 @@ import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.dsl.channel.MessageChannelSpec;
import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
import org.springframework.integration.dsl.support.BeanNameMethodInvokingMessageHandler;
import org.springframework.integration.dsl.support.EndpointConfigurer;
import org.springframework.integration.filter.ExpressionEvaluatingSelector;
import org.springframework.integration.filter.MessageFilter;
@@ -68,7 +69,7 @@ public final class IntegrationFlowBuilder {
public IntegrationFlowBuilder channel(MessageChannel messageChannel) {
Assert.notNull(messageChannel);
if (this.currentMessageChannel != null) {
GenericEndpointSpec endpointSpec = new GenericEndpointSpec(new BridgeHandler());
GenericEndpointSpec<BridgeHandler> endpointSpec = new GenericEndpointSpec<BridgeHandler>(new BridgeHandler());
endpointSpec.get().getT1().setInputChannel(this.currentMessageChannel);
this.addComponent(endpointSpec).currentComponent(endpointSpec.get().getT2());
}
@@ -90,7 +91,7 @@ public final class IntegrationFlowBuilder {
}
public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer,
EndpointConfigurer<GenericEndpointSpec> endpointConfigurer) {
EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
Transformer transformer = genericTransformer instanceof Transformer
? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer);
return this.handle(new MessageTransformingHandler(transformer), endpointConfigurer);
@@ -114,12 +115,20 @@ public final class IntegrationFlowBuilder {
return this.handle(messageHandler, null);
}
public IntegrationFlowBuilder handle(MessageHandler messageHandler, EndpointConfigurer<GenericEndpointSpec> endpointConfigurer) {
return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer);
public IntegrationFlowBuilder handle(String target, String methodName) {
return this.handle(target, methodName, null);
}
public IntegrationFlowBuilder bridge(EndpointConfigurer<GenericEndpointSpec> endpointConfigurer) {
return this.register(new GenericEndpointSpec(new BridgeHandler()), endpointConfigurer);
public IntegrationFlowBuilder handle(String beanName, String methodName, EndpointConfigurer<GenericEndpointSpec<BeanNameMethodInvokingMessageHandler>> endpointConfigurer) {
return this.handle(new BeanNameMethodInvokingMessageHandler(beanName, methodName) , endpointConfigurer);
}
public <H extends MessageHandler> IntegrationFlowBuilder handle(H messageHandler, EndpointConfigurer<GenericEndpointSpec<H>> endpointConfigurer) {
return this.register(new GenericEndpointSpec<H>(messageHandler), endpointConfigurer);
}
public IntegrationFlowBuilder bridge(EndpointConfigurer<GenericEndpointSpec<BridgeHandler>> endpointConfigurer) {
return this.register(new GenericEndpointSpec<BridgeHandler>(new BridgeHandler()), endpointConfigurer);
}
private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) {

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java
View File

@@ -7,7 +7,7 @@ import org.springframework.integration.scheduling.PollerMetadata;
/**
* @author Artem Bilan
* @since 4.0
*/
public final class SourcePollingChannelAdapterSpec
extends EndpointSpec<SourcePollingChannelAdapterSpec, SourcePollingChannelAdapterFactoryBean, MessageSource<?>> {

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/ConsumerEndpointSpec.java
View File

@@ -29,7 +29,7 @@ import org.springframework.messaging.MessageHandler;
/**
* @author Artem Bilan
* @since 4.0
*/
public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>, H extends MessageHandler>
extends EndpointSpec<S, ConsumerEndpointFactoryBean, H> {

20
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java
View File

@@ -47,7 +47,8 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig
public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
Assert.isInstanceOf(BeanDefinitionRegistry.class, configurableListableBeanFactory,
"To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." +
"Consider using 'GenericApplicationContext' implementation.");
"Consider using 'GenericApplicationContext' implementation."
);
this.initializeIntegrationFlows(configurableListableBeanFactory);
this.populateBeansFromSpecs(configurableListableBeanFactory);
}
@@ -78,10 +79,15 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig
ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1();
String id = endpointSpec.getId();
String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler);
String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null;
BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias);
BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry);
Collection<?> messageHandlers = beanFactory.getBeansOfType(messageHandler.getClass(), false, false).values();
if (!messageHandlers.contains(messageHandler)) {
String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler);
String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null;
BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler),
handlerBeanName, handlerAlias);
BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry);
}
String endpointBeanName = id;
if (endpointBeanName == null) {
@@ -111,9 +117,7 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig
for (Map.Entry<String, ?> specEntry : specs.entrySet()) {
String id = specEntry.getKey();
IntegrationComponentSpec<?, ?> spec = (IntegrationComponentSpec<?, ?>) specEntry.getValue();
registry.removeBeanDefinition(id);
beanFactory.registerSingleton(id, spec.get());
beanFactory.initializeBean(spec.get(), id);
registry.registerBeanDefinition(id, new InstanceBeanDefinition(spec.get()));
}
}

6
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java
View File

@@ -1,7 +1,7 @@
package org.springframework.integration.dsl.core;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.core.GenericTypeResolver;
import org.springframework.core.ResolvableType;
import org.springframework.integration.dsl.support.PollerSpec;
import org.springframework.integration.dsl.tuple.Tuple;
import org.springframework.integration.dsl.tuple.Tuple2;
@@ -9,14 +9,14 @@ import org.springframework.integration.scheduling.PollerMetadata;
/**
* @author Artem Bilan
* @since 4.0
*/
public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends BeanNameAware, H> extends IntegrationComponentSpec<S, Tuple2<F, H>> {
@SuppressWarnings("unchecked")
protected EndpointSpec(H handler) {
try {
Class<?> fClass = GenericTypeResolver.resolveTypeArguments(this.getClass(), EndpointSpec.class)[1];
Class<?> fClass = ResolvableType.forClass(this.getClass()).as(EndpointSpec.class).resolveGenerics()[1];
F endpointFactoryBean = (F) fClass.newInstance();
this.target = Tuple.of(endpointFactoryBean, handler);
}

4
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationComponentSpec.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.dsl.core;
/**
* @author Artem Bilan
* @since 4.0
*/
public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpec<S, T>, T> {
@@ -31,7 +31,7 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
return _this();
}
String getId() {
final String getId() {
return id;
}

35
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/BeanNameMethodInvokingMessageHandler.java Normal file
View File

@@ -0,0 +1,35 @@
package org.springframework.integration.dsl.support;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.handler.MethodInvokingMessageProcessor;
import org.springframework.messaging.Message;
/**
* @author Artem Bilan
*/
public final class BeanNameMethodInvokingMessageHandler extends AbstractReplyProducingMessageHandler {
private final String object;
private final String methodName;
private MessageProcessor<Object> processor;
public BeanNameMethodInvokingMessageHandler(String object, String methodName) {
this.object = object;
this.methodName = methodName;
}
@Override
protected void doInit() {
Object target = this.getBeanFactory().getBean(object);
this.processor = new MethodInvokingMessageProcessor<Object>(target, this.methodName);
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return this.processor.processMessage(requestMessage);
}
}

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java
View File

@@ -20,7 +20,7 @@ import org.springframework.integration.dsl.core.EndpointSpec;
/**
* @author Artem Bilan
* @since 4.0
*/
public interface EndpointConfigurer<S extends EndpointSpec<?, ?, ?>> {

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java
View File

@@ -34,7 +34,7 @@ import org.springframework.util.ErrorHandler;
/**
* @author Artem Bilan
* @since 4.0
*/
public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, PollerMetadata> {

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

@@ -23,6 +23,7 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.File;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -32,10 +33,9 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -59,6 +59,9 @@ import org.springframework.integration.dsl.support.Pollers;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.event.core.MessagingEvent;
import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
import org.springframework.integration.file.DefaultFileNameGenerator;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.store.SimpleMessageStore;
@@ -67,8 +70,11 @@ import org.springframework.integration.transformer.PayloadDeserializingTransform
import org.springframework.integration.transformer.PayloadSerializingTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.support.AnnotationConfigContextLoader;
@@ -80,8 +86,10 @@ import org.springframework.test.context.support.AnnotationConfigContextLoader;
@RunWith(SpringJUnit4ClassRunner.class)
public class IntegrationFlowTests {
private static final File tmpDir = new File(System.getProperty("java.io.tmpdir"));
@Autowired
private BeanFactory beanFactory;
private ListableBeanFactory beanFactory;
@Autowired
@Qualifier("flow1QueueChannel")
@@ -118,6 +126,18 @@ public class IntegrationFlowTests {
@Qualifier("bridgeFlow2Output")
private PollableChannel bridgeFlow2Output;
@Autowired
@Qualifier("fileFlow1Input")
private DirectChannel fileFlow1Input;
@Autowired
@Qualifier("fileWritingMessageHandler")
private FileWritingMessageHandler fileWritingMessageHandler;
@Autowired
@Qualifier("methodInvokingInput")
private DirectChannel methodInvokingInput;
@Test
public void testPollingFlow() {
for (int i = 0; i < 10; i++) {
@@ -197,6 +217,34 @@ public class IntegrationFlowTests {
}
@Test
public void testFileHandler() {
assertEquals(1, this.beanFactory.getBeansOfType(FileWritingMessageHandler.class).size());
Message<?> message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build();
try {
this.fileFlow1Input.send(message);
fail("NullPointerException expected");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(MessageHandlingException.class));
assertThat(e.getCause(), Matchers.instanceOf(NullPointerException.class));
}
this.fileWritingMessageHandler.setFileNameGenerator(new DefaultFileNameGenerator());
this.fileFlow1Input.send(message);
assertTrue(new File(tmpDir, "foo").exists());
}
@Test
public void testMethodInvokingMessageHandler() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message = MessageBuilder.withPayload("world").setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel).build();
this.methodInvokingInput.send(message);
Message<?> receive = replyChannel.receive(5000);
assertNotNull(receive);
assertEquals("Hello, world", receive.getPayload());
}
@Configuration
@EnableIntegration
@@ -315,6 +363,41 @@ public class IntegrationFlowTests {
}
@Configuration
public static class ContextConfiguration4 {
@Bean
public FileWritingMessageHandler fileWritingMessageHandler() {
return new FileWritingMessageHandler(tmpDir);
}
@Bean
public IntegrationFlow fileFlow1() {
return IntegrationFlows.from(MessageChannels.direct("fileFlow1Input"))
.handle(this.fileWritingMessageHandler(), c -> {
FileWritingMessageHandler handler = c.get().getT2();
handler.setFileNameGenerator(message -> null);
handler.setExpectReply(false); })
.get();
}
@Bean
public IntegrationFlow methodInvokingFlow() {
return IntegrationFlows.from(MessageChannels.direct("methodInvokingInput"))
.handle("greetingService", null)
.get();
}
}
@Component("greetingService")
public static class GreetingService {
public String greeting(String payload) {
return "Hello, " + payload;
}
}
private static class InvalidLastComponentFlowContext {
@Bean