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

DSL: Fixes and polishing after upgrade

Browse Source 
* Upgrade to SI 4.0.3 and Boot 1.1.4
* Add `HeaderEnricherSpec#headerChannelsToString()`
* Fix `DslRecipientListRouter` according core changes
* Fix `Amqp` and `Jms` modules for the bug around "All Message creators need a BeanFactory"
using fake `new DefaultListableBeanFactory()` for `(Amqp)JmsChannelFactoryBean`
This commit is contained in:
Artem Bilan
2014-07-17 14:53:17 +03:00
parent 24e8c59878
commit f6c9f1305b
9 changed files with 57 additions and 35 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/DslRecipientListRouter.java
View File

@@ -55,7 +55,7 @@ class DslRecipientListRouter extends RecipientListRouter {
}
@Override
public void onInit() {
public void onInit() throws Exception {
for (Map.Entry<String, String> recipient : this.expressionRecipientMap.entrySet()) {
ExpressionEvaluatingSelector selector = null;
String expression = recipient.getValue();

12
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java
View File

@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dsl.core.IntegrationComponentSpec;
import org.springframework.integration.dsl.support.BeanNameMessageProcessor;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
@@ -97,6 +98,17 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
return _this();
}
public <V> HeaderEnricherSpec headerChannelsToString() {
return headerExpression("replyChannel",
"@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME
+ ".channelToChannelName(headers.replyChannel)",
true)
.headerExpression("errorChannel",
"@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME
+ ".channelToChannelName(headers.errorChannel)",
true);
}
@Override
protected HeaderEnricher doGet() {
return this.headerEnricher;

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

@@ -20,6 +20,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
import org.springframework.integration.dsl.channel.MessageChannelSpec;
@@ -42,6 +43,7 @@ public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannel
this.amqpChannelFactoryBean.setConnectionFactory(connectionFactory);
this.amqpChannelFactoryBean.setSingleton(false);
this.amqpChannelFactoryBean.setPubSub(false);
this.amqpChannelFactoryBean.setBeanFactory(new DefaultListableBeanFactory());
}
@Override

1
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGateway.java
View File

@@ -87,6 +87,7 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
super.setApplicationContext(applicationContext);
this.endpoint.setApplicationContext(applicationContext);
this.endpoint.setBeanFactory(applicationContext);
this.listener.setBeanFactory(applicationContext);
}

1
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageChannelSpec.java
View File

@@ -128,6 +128,7 @@ public class JmsMessageChannelSpec<S extends JmsMessageChannelSpec<S>> extends J
protected AbstractJmsChannel doGet() {
AbstractJmsChannel jmsChannel = super.doGet();
if (this.cacheLevel != null) {
//TODO till INT-3435
DirectFieldAccessor dfa = new DirectFieldAccessor(jmsChannel);
Object container = dfa.getPropertyValue("container");
if (container instanceof DefaultMessageListenerContainer) {

1
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapter.java
View File

@@ -77,6 +77,7 @@ public class JmsMessageDrivenChannelAdapter extends MessageProducerSupport imple
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
super.setApplicationContext(applicationContext);
this.endpoint.setApplicationContext(applicationContext);
this.endpoint.setBeanFactory(applicationContext);
this.listener.setBeanFactory(applicationContext);
}

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

@@ -20,6 +20,7 @@ import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.integration.dsl.channel.MessageChannelSpec;
import org.springframework.integration.jms.AbstractJmsChannel;
import org.springframework.integration.jms.config.JmsChannelFactoryBean;
@@ -42,6 +43,7 @@ public class JmsPollableMessageChannelSpec<S extends JmsPollableMessageChannelSp
this.jmsChannelFactoryBean = jmsChannelFactoryBean;
this.jmsChannelFactoryBean.setConnectionFactory(connectionFactory);
this.jmsChannelFactoryBean.setSingleton(false);
this.jmsChannelFactoryBean.setBeanFactory(new DefaultListableBeanFactory());
}
@Override

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

@@ -16,6 +16,7 @@
package org.springframework.integration.dsl.test;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;
import java.io.File;
@@ -330,7 +331,7 @@ public class IntegrationFlowTests {
@Test
public void testPollingFlow() {
this.controlBus.send("@integerEndpoint.start()");
assertThat(this.beanFactory.getBean("integerChannel"), Matchers.instanceOf(FixedSubscriberChannel.class));
assertThat(this.beanFactory.getBean("integerChannel"), instanceOf(FixedSubscriberChannel.class));
for (int i = 0; i < 5; i++) {
Message<?> message = this.outputChannel.receive(20000);
assertNotNull(message);
@@ -354,9 +355,9 @@ public class IntegrationFlowTests {
fail("Expected MessageDispatchingException");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), containsString("Dispatcher has no subscribers"));
}
this.controlBus.send("@payloadSerializingTransformer.start()");
@@ -393,7 +394,7 @@ public class IntegrationFlowTests {
assertEquals("test", reply.getPayload());
assertTrue(this.beanFactory.containsBean("bridgeFlow2.channel#0"));
assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), Matchers.instanceOf(FixedSubscriberChannel
assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), instanceOf(FixedSubscriberChannel
.class));
try {
@@ -401,9 +402,9 @@ public class IntegrationFlowTests {
fail("Expected MessageDispatchingException");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), containsString("Dispatcher has no subscribers"));
}
this.controlBus.send("@bridge.start()");
this.bridgeFlow2Input.send(message);
@@ -421,8 +422,8 @@ public class IntegrationFlowTests {
fail("BeanCreationException expected");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(BeanCreationException.class));
assertThat(e.getMessage(), Matchers.containsString("is a one-way 'MessageHandler'"));
assertThat(e, instanceOf(BeanCreationException.class));
assertThat(e.getMessage(), containsString("is a one-way 'MessageHandler'"));
}
finally {
if (context != null) {
@@ -439,8 +440,8 @@ public class IntegrationFlowTests {
fail("BeanCreationException expected");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(BeanCreationException.class));
assertThat(e.getMessage(), Matchers.containsString("'.fixedSubscriberChannel()' " +
assertThat(e, instanceOf(BeanCreationException.class));
assertThat(e.getMessage(), containsString("'.fixedSubscriberChannel()' " +
"can't be the last EIP-method in the IntegrationFlow definition"));
}
finally {
@@ -459,8 +460,8 @@ public class IntegrationFlowTests {
fail("NullPointerException expected");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(MessageHandlingException.class));
assertThat(e.getCause(), Matchers.instanceOf(NullPointerException.class));
assertThat(e, instanceOf(MessageHandlingException.class));
assertThat(e.getCause(), instanceOf(NullPointerException.class));
}
DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
fileNameGenerator.setBeanFactory(this.beanFactory);
@@ -518,10 +519,10 @@ public class IntegrationFlowTests {
fail("BeanCreationException expected");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(IllegalArgumentException.class));
assertThat(e.getCause(), Matchers.instanceOf(BeanCreationException.class));
assertThat(e, instanceOf(IllegalArgumentException.class));
assertThat(e.getCause(), instanceOf(BeanCreationException.class));
assertThat(e.getCause().getMessage(),
Matchers.containsString("must be populated to target objects via 'get()' method call"));
containsString("must be populated to target objects via 'get()' method call"));
}
finally {
if (context != null) {
@@ -541,7 +542,7 @@ public class IntegrationFlowTests {
assertNotNull(receive);
assertEquals("Bar Bar", receive.getHeaders().get("foo"));
Object payload = receive.getPayload();
assertThat(payload, Matchers.instanceOf(TestPojo.class));
assertThat(payload, instanceOf(TestPojo.class));
TestPojo result = (TestPojo) payload;
assertEquals("Bar Bar", result.getName());
assertNotNull(result.getDate());
@@ -578,7 +579,7 @@ public class IntegrationFlowTests {
Message<?> receive = replyChannel.receive(2000);
assertNotNull(receive);
assertThat(receive.getPayload(), Matchers.instanceOf(List.class));
assertThat(receive.getPayload(), instanceOf(List.class));
@SuppressWarnings("unchecked")
List<Object> result = (List<Object>) receive.getPayload();
for (int i = 0; i < payload.size(); i++) {
@@ -600,9 +601,9 @@ public class IntegrationFlowTests {
fail("Expected MessageDispatchingException");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(MessageDispatchingException.class));
assertThat(e.getMessage(), containsString("Dispatcher has no subscribers"));
}
this.controlBus.send("@xpathHeaderEnricher.start()");
@@ -613,6 +614,7 @@ public class IntegrationFlowTests {
MessageHeaders headers = result.getHeaders();
assertEquals("1", headers.get("one"));
assertEquals("2", headers.get("two"));
assertThat(headers.getReplyChannel(), instanceOf(String.class));
}
@Test
@@ -661,7 +663,7 @@ public class IntegrationFlowTests {
}
catch (MessageDeliveryException e) {
assertThat(e.getMessage(),
Matchers.containsString("no channel resolved by router and no default output channel defined"));
containsString("no channel resolved by router and no default output channel defined"));
}
}
@@ -690,9 +692,9 @@ public class IntegrationFlowTests {
fail("DestinationResolutionException expected.");
}
catch (MessagingException e) {
assertThat(e.getCause(), Matchers.instanceOf(DestinationResolutionException.class));
assertThat(e.getCause(), instanceOf(DestinationResolutionException.class));
assertThat(e.getCause().getMessage(),
Matchers.containsString("failed to look up MessageChannel with name 'bad-channel'"));
containsString("failed to look up MessageChannel with name 'bad-channel'"));
}
}
@@ -721,9 +723,9 @@ public class IntegrationFlowTests {
fail("DestinationResolutionException expected.");
}
catch (MessagingException e) {
assertThat(e.getCause(), Matchers.instanceOf(DestinationResolutionException.class));
assertThat(e.getCause(), instanceOf(DestinationResolutionException.class));
assertThat(e.getCause().getMessage(),
Matchers.containsString("failed to look up MessageChannel with name 'bad-channel'"));
containsString("failed to look up MessageChannel with name 'bad-channel'"));
}
}
@@ -756,7 +758,7 @@ public class IntegrationFlowTests {
}
catch (MessageDeliveryException e) {
assertThat(e.getMessage(),
Matchers.containsString("no channel resolved by router and no default output channel defined"));
containsString("no channel resolved by router and no default output channel defined"));
}
}
@@ -908,9 +910,9 @@ public class IntegrationFlowTests {
receive = this.gatewayError.receive(2000);
assertNotNull(receive);
assertThat(receive, Matchers.instanceOf(ErrorMessage.class));
assertThat(receive.getPayload(), Matchers.instanceOf(MessageRejectedException.class));
assertThat(((Exception) receive.getPayload()).getMessage(), Matchers.containsString("' rejected Message"));
assertThat(receive, instanceOf(ErrorMessage.class));
assertThat(receive.getPayload(), instanceOf(MessageRejectedException.class));
assertThat(((Exception) receive.getPayload()).getMessage(), containsString("' rejected Message"));
}
@Autowired
@@ -1399,7 +1401,8 @@ public class IntegrationFlowTests {
return IntegrationFlows.from("xpathHeaderEnricherInput")
.enrichHeaders(
s -> s.header("one", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne"))
.header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo")),
.header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo"))
.headerChannelsToString(),
c -> c.autoStartup(false).id("xpathHeaderEnricher")
)
.get();