From dae1a010038215fb676a0fb68f33a84783a4fefa Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 29 May 2016 11:57:22 -0400 Subject: [PATCH] INT-4043: Fix ExecutorChannel with datatypes Attr. JIRA: https://jira.spring.io/browse/INT-4043 The `ExecutorChannel` overrides `onInit()` but fails to call the super which is where the message converter for datatype conversion is set up. Also, when Jackson is not on the class path and there are no converters in the context, the default integration conversion service is not registered. The `DefaultDatatypeChannelMessageConverter` overwites its default conversion service with this bean, unconditionally - setting it to null in this case. Check for a null conversion service before replacing the default. * Polishing according PR comments --- .../integration/channel/ExecutorChannel.java | 8 +- ...efaultDatatypeChannelMessageConverter.java | 5 +- ...sts.xml => ChannelParserTests-context.xml} | 10 +- ...hannelParserTests-no-converter-context.xml | 16 +++ .../channel/config/ChannelParserTests.java | 127 +++++++++--------- 5 files changed, 95 insertions(+), 71 deletions(-) rename spring-integration-core/src/test/java/org/springframework/integration/channel/config/{channelParserTests.xml => ChannelParserTests-context.xml} (81%) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-no-converter-context.xml diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java index f7ec4f349b..a3730f914c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -101,6 +101,12 @@ public class ExecutorChannel extends AbstractExecutorChannel { @Override public final void onInit() { + try { + super.onInit(); // TODO add throws clause in 5.0 + } + catch (Exception e) { + throw new IllegalStateException(e); + } if (!(this.executor instanceof ErrorHandlingTaskExecutor)) { ErrorHandler errorHandler = new MessagePublishingErrorHandler( new BeanFactoryChannelResolver(this.getBeanFactory())); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java index 877680a0ba..654820365e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java @@ -60,7 +60,10 @@ public class DefaultDatatypeChannelMessageConverter implements MessageConverter, @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { if (!this.conversionServiceSet && beanFactory != null) { - this.conversionService = IntegrationUtils.getConversionService(beanFactory); + ConversionService integrationConversionService = IntegrationUtils.getConversionService(beanFactory); + if (integrationConversionService != null) { + this.conversionService = integrationConversionService; + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-context.xml similarity index 81% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml rename to spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-context.xml index 0f68088a85..3071b34203 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-context.xml @@ -35,7 +35,8 @@ - + @@ -46,7 +47,10 @@ - + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-no-converter-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-no-converter-context.xml new file mode 100644 index 0000000000..1f2dee5540 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-no-converter-context.xml @@ -0,0 +1,16 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index bb734e8476..2211d24d90 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,13 +29,17 @@ import static org.junit.Assert.assertTrue; import java.util.concurrent.Executor; import org.junit.Test; +import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.FatalBeanException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.convert.converter.Converter; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ExecutorChannel; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.TestChannelInterceptor; @@ -53,6 +57,9 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher @@ -62,8 +69,16 @@ import org.springframework.messaging.support.GenericMessage; * * @see ChannelWithCustomQueueParserTests */ +@ContextConfiguration(locations = { + "/org/springframework/integration/channel/config/ChannelParserTests-context.xml", + "/org/springframework/integration/channel/config/priorityChannelParserTests.xml" }) +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class ChannelParserTests { + @Autowired + private ApplicationContext context; + @Test(expected = FatalBeanException.class) public void testChannelWithoutId() { new ClassPathXmlApplicationContext("channelWithoutId.xml", this.getClass()).close(); @@ -71,56 +86,62 @@ public class ChannelParserTests { @Test public void testChannelWithCapacity() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("capacityChannel"); for (int i = 0; i < 10; i++) { boolean result = channel.send(new GenericMessage("test"), 10); assertTrue(result); } assertFalse(channel.send(new GenericMessage("test"), 3)); - context.close(); } @Test public void testDirectChannelByDefault() throws InterruptedException { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("defaultChannel"); - assertEquals(DirectChannel.class, channel.getClass()); + assertThat(channel, instanceOf(DirectChannel.class)); DirectFieldAccessor accessor = new DirectFieldAccessor(channel); Object dispatcher = accessor.getPropertyValue("dispatcher"); assertThat(dispatcher, is(instanceOf(UnicastingDispatcher.class))); assertThat(new DirectFieldAccessor(dispatcher).getPropertyValue("loadBalancingStrategy"), is(instanceOf(RoundRobinLoadBalancingStrategy.class))); + } + + @Test + public void testExecutorChannel() throws InterruptedException { + MessageChannel channel = context.getBean("executorChannel", MessageChannel.class); + assertThat(channel, instanceOf(ExecutorChannel.class)); + assertNotNull(TestUtils.getPropertyValue(channel, "messageConverter")); + assertNotNull(TestUtils.getPropertyValue(channel, "messageConverter.conversionService")); + } + + @Test + public void testExecutorChannelNoConverter() throws InterruptedException { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "ChannelParserTests-no-converter-context.xml", this.getClass()); + MessageChannel channel = context.getBean("executorChannel", MessageChannel.class); + assertThat(channel, instanceOf(ExecutorChannel.class)); + assertNotNull(TestUtils.getPropertyValue(channel, "messageConverter")); + assertNotNull(TestUtils.getPropertyValue(channel, "messageConverter.conversionService")); context.close(); } @Test public void channelWithFailoverDispatcherAttribute() throws Exception { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("channelWithFailover"); assertEquals(DirectChannel.class, channel.getClass()); DirectFieldAccessor accessor = new DirectFieldAccessor(channel); Object dispatcher = accessor.getPropertyValue("dispatcher"); assertThat(dispatcher, is(instanceOf(UnicastingDispatcher.class))); assertNull(new DirectFieldAccessor(dispatcher).getPropertyValue("loadBalancingStrategy")); - context.close(); } @Test public void testPublishSubscribeChannel() throws InterruptedException { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel"); assertEquals(PublishSubscribeChannel.class, channel.getClass()); } @Test public void testPublishSubscribeChannelWithTaskExecutorReference() throws InterruptedException { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannelWithTaskExecutorRef"); assertEquals(PublishSubscribeChannel.class, channel.getClass()); DirectFieldAccessor accessor = new DirectFieldAccessor(channel); @@ -132,44 +153,34 @@ public class ChannelParserTests { Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor"); Object executorBean = context.getBean("taskExecutor"); assertEquals(executorBean, innerExecutor); - context.close(); } @Test public void channelWithCustomQueue() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "channelParserTests.xml", this.getClass()); Object customQueue = context.getBean("customQueue"); Object channelWithCustomQueue = context.getBean("channelWithCustomQueue"); assertEquals(QueueChannel.class, channelWithCustomQueue.getClass()); Object actualQueue = new DirectFieldAccessor(channelWithCustomQueue).getPropertyValue("queue"); assertSame(customQueue, actualQueue); - context.close(); } @Test public void testDatatypeChannelWithCorrectType() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("integerChannel"); assertTrue(channel.send(new GenericMessage(123))); - context.close(); } @Test(expected = MessageDeliveryException.class) public void testDatatypeChannelWithIncorrectType() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("integerChannel"); channel.send(new GenericMessage("incorrect type")); - context.close(); assertTrue(TestUtils.getPropertyValue(channel, "messageConverter") instanceof UselessMessageConverter); } @Test public void testDatatypeChannelGlobalConverter() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserGlobalConverterTests.xml", this - .getClass()); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("channelParserGlobalConverterTests.xml", getClass()); MessageChannel channel = context.getBean("integerChannel", MessageChannel.class); context.close(); assertTrue(TestUtils.getPropertyValue(channel, "messageConverter") instanceof UselessMessageConverter); @@ -177,44 +188,36 @@ public class ChannelParserTests { @Test public void testDatatypeChannelWithAssignableSubTypes() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("numberChannel"); - assertTrue(channel.send(new GenericMessage(123))); - assertTrue(channel.send(new GenericMessage(123.45))); - assertTrue(channel.send(new GenericMessage(Boolean.TRUE))); - assertTrue(TestUtils.getPropertyValue(channel, "messageConverter") instanceof DefaultDatatypeChannelMessageConverter); + assertTrue(channel.send(new GenericMessage<>(123))); + assertTrue(channel.send(new GenericMessage<>(123.45))); + assertTrue(channel.send(new GenericMessage<>(Boolean.TRUE))); + assertThat(TestUtils.getPropertyValue(channel, "messageConverter"), + instanceOf(DefaultDatatypeChannelMessageConverter.class)); assertNotNull(TestUtils.getPropertyValue(channel, "messageConverter.conversionService")); - context.close(); } @Test public void testMultipleDatatypeChannelWithCorrectTypes() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel"); - assertTrue(channel.send(new GenericMessage(123))); - assertTrue(channel.send(new GenericMessage("accepted type"))); - context.close(); + assertTrue(channel.send(new GenericMessage<>(123))); + assertTrue(channel.send(new GenericMessage<>("accepted type"))); } @Test(expected = MessageDeliveryException.class) public void testMultipleDatatypeChannelWithIncorrectType() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("channelParserTests.xml", this - .getClass()); MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel"); - channel.send(new GenericMessage(Boolean.TRUE)); - context.close(); + channel.send(new GenericMessage<>(Boolean.TRUE)); } @Test public void testChannelInteceptorRef() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", this - .getClass()); + ConfigurableApplicationContext context = + new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", getClass()); PollableChannel channel = (PollableChannel) context.getBean("channelWithInterceptorRef"); TestChannelInterceptor interceptor = (TestChannelInterceptor) context.getBean("interceptor"); assertEquals(0, interceptor.getSendCount()); - channel.send(new GenericMessage("test")); + channel.send(new GenericMessage<>("test")); assertEquals(1, interceptor.getSendCount()); assertEquals(0, interceptor.getReceiveCount()); channel.receive(); @@ -224,8 +227,8 @@ public class ChannelParserTests { @Test public void testChannelInteceptorInnerBean() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", this - .getClass()); + ConfigurableApplicationContext context = + new ClassPathXmlApplicationContext("channelInterceptorParserTests.xml", getClass()); PollableChannel channel = (PollableChannel) context.getBean("channelWithInterceptorInnerBean"); channel.send(new GenericMessage("test")); Message transformed = channel.receive(1000); @@ -235,8 +238,7 @@ public class ChannelParserTests { @Test public void testPriorityChannelWithDefaultComparator() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this.getClass()); - PollableChannel channel = (PollableChannel) context.getBean("priorityChannelWithDefaultComparator"); + PollableChannel channel = this.context.getBean("priorityChannelWithDefaultComparator", PollableChannel.class); Message lowPriorityMessage = MessageBuilder.withPayload("low").setPriority(-14).build(); Message midPriorityMessage = MessageBuilder.withPayload("mid").setPriority(0).build(); Message highPriorityMessage = MessageBuilder.withPayload("high").setPriority(99).build(); @@ -249,18 +251,15 @@ public class ChannelParserTests { assertEquals("high", reply1.getPayload()); assertEquals("mid", reply2.getPayload()); assertEquals("low", reply3.getPayload()); - context.close(); } @Test public void testPriorityChannelWithCustomComparator() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this - .getClass()); - PollableChannel channel = (PollableChannel) context.getBean("priorityChannelWithCustomComparator"); - channel.send(new GenericMessage("C")); - channel.send(new GenericMessage("A")); - channel.send(new GenericMessage("D")); - channel.send(new GenericMessage("B")); + PollableChannel channel = this.context.getBean("priorityChannelWithCustomComparator", PollableChannel.class); + channel.send(new GenericMessage<>("C")); + channel.send(new GenericMessage<>("A")); + channel.send(new GenericMessage<>("D")); + channel.send(new GenericMessage<>("B")); Message reply1 = channel.receive(0); Message reply2 = channel.receive(0); Message reply3 = channel.receive(0); @@ -269,30 +268,26 @@ public class ChannelParserTests { assertEquals("B", reply2.getPayload()); assertEquals("C", reply3.getPayload()); assertEquals("D", reply4.getPayload()); - context.close(); } @Test public void testPriorityChannelWithIntegerDatatypeEnforced() { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext("priorityChannelParserTests.xml", this - .getClass()); - PollableChannel channel = (PollableChannel) context.getBean("integerOnlyPriorityChannel"); - channel.send(new GenericMessage(3)); - channel.send(new GenericMessage(2)); - channel.send(new GenericMessage(1)); + PollableChannel channel = this.context.getBean("integerOnlyPriorityChannel", PollableChannel.class); + channel.send(new GenericMessage<>(3)); + channel.send(new GenericMessage<>(2)); + channel.send(new GenericMessage<>(1)); assertEquals(1, channel.receive(0).getPayload()); assertEquals(2, channel.receive(0).getPayload()); assertEquals(3, channel.receive(0).getPayload()); boolean threwException = false; try { - channel.send(new GenericMessage("wrong type")); + channel.send(new GenericMessage<>("wrong type")); } catch (MessageDeliveryException e) { assertEquals("wrong type", e.getFailedMessage().getPayload()); threwException = true; } assertTrue(threwException); - context.close(); } public static class TestInterceptor extends ChannelInterceptorAdapter {