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

INT-4043: Fix ExecutorChannel with datatypes Attr.

Browse Source 
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
This commit is contained in:
Gary Russell
2016-05-29 11:57:22 -04:00
committed by Artem Bilan
parent 17abf29da1
commit dae1a01003
5 changed files with 95 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -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()));

5
spring-integration-core/src/main/java/org/springframework/integration/support/converter/DefaultDatatypeChannelMessageConverter.java
View File

@@ -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;
}
}
}

10
spring-integration-core/src/test/java/org/springframework/integration/channel/config/channelParserTests.xml → spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-context.xml
View File

@@ -35,7 +35,8 @@
<queue capacity="10" />
</channel>
<beans:bean id="uselessConverter" class="org.springframework.integration.channel.config.ChannelParserTests$UselessMessageConverter" />
<beans:bean id="uselessConverter"
class="org.springframework.integration.channel.config.ChannelParserTests$UselessMessageConverter" />
<channel id="numberChannel" datatype="java.lang.Number">
<queue capacity="10" />
@@ -46,7 +47,10 @@
<queue capacity="10" />
</channel>
<beans:bean id="taskExecutor"
class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
<channel id="executorChannel" datatype="java.lang.Byte[]">
<dispatcher task-executor="taskExecutor" />
</channel>
<beans:bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
</beans:beans>

16
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests-no-converter-context.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="executorChannel" datatype="[B">
<dispatcher task-executor="taskExecutor" />
</channel>
<beans:bean id="taskExecutor"
class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
</beans:beans>

127
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java
View File

@@ -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<String>("test"), 10);
assertTrue(result);
}
assertFalse(channel.send(new GenericMessage<String>("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<Integer>(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<String>("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<Integer>(123)));
assertTrue(channel.send(new GenericMessage<Double>(123.45)));
assertTrue(channel.send(new GenericMessage<Boolean>(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<Integer>(123)));
assertTrue(channel.send(new GenericMessage<String>("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>(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<String>("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<String>("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<String> lowPriorityMessage = MessageBuilder.withPayload("low").setPriority(-14).build();
Message<String> midPriorityMessage = MessageBuilder.withPayload("mid").setPriority(0).build();
Message<String> 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<String>("C"));
channel.send(new GenericMessage<String>("A"));
channel.send(new GenericMessage<String>("D"));
channel.send(new GenericMessage<String>("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<Integer>(3));
channel.send(new GenericMessage<Integer>(2));
channel.send(new GenericMessage<Integer>(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<String>("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 {