diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java index d4533086fe..79eb26063b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java @@ -49,6 +49,7 @@ public abstract class AbstractInboundGatewayParser extends AbstractSimpleBeanDef @Override protected boolean isEligibleAttribute(String attributeName) { return !attributeName.equals("name") && !attributeName.equals("request-channel") + && !attributeName.equals("error-channel") && !attributeName.equals("reply-channel") && super.isEligibleAttribute(attributeName); } @@ -61,6 +62,10 @@ public abstract class AbstractInboundGatewayParser extends AbstractSimpleBeanDef if (StringUtils.hasText(replyChannel)) { builder.addPropertyReference("replyChannel", replyChannel); } + String errorChannel = element.getAttribute("error-channel"); + if (StringUtils.hasText(errorChannel)) { + builder.addPropertyReference("errorChannel", errorChannel); + } String autoStartup = element.getAttribute("auto-startup"); if (StringUtils.hasText(autoStartup)) { builder.addPropertyValue("autoStartup", autoStartup); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java index b49130f60e..bae873546c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java @@ -40,6 +40,8 @@ public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser IpAdapterParserUtils.TCP_CONNECTION_FACTORY); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, + element, "error-channel", "errorChannel"); return builder.getBeanDefinition(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java index 88ff3a942b..80830b3294 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java @@ -43,6 +43,8 @@ public class UdpInboundChannelAdapterParser extends AbstractChannelAdapterParser IpAdapterParserUtils.POOL_SIZE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, + element, "error-channel", "errorChannel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, IpAdapterParserUtils.TASK_EXECUTOR); return builder.getBeanDefinition(); diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 12dd334a58..bef826260f 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -48,6 +48,15 @@ its configuration specifies the number of threads. + + + + + + + + + @@ -110,6 +119,15 @@ adapter. + + + + + + + + + @@ -177,6 +195,15 @@ A connection factory is needed by an inbound adapter. The connection factory mus + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index e739ba9dd6..1b888f7d9d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -29,6 +29,7 @@ so-timeout="32" local-address="127.0.0.1" task-executor="externalTE" + error-channel="errorChannel" /> @@ -105,13 +107,26 @@ port="#{tcpIpUtils.findAvailableServerSocket(5800)}" /> - + + + + message) throws MessagingException { + MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel(); + replyChannel.send(new GenericMessage(errorMessage)); + } + }); + gateway.setErrorChannel(errorChannel); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 200) { + fail("Failed to listen"); + } + } + final SubscribableChannel channel = new DirectChannel(); + gateway.setRequestChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService()); + channel.subscribe(handler); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test1\r\n".getBytes()); + socket.getOutputStream().write("Test2\r\n".getBytes()); + byte[] bytes = new byte[errorMessage.length() + 2]; + readFully(socket.getInputStream(), bytes); + assertEquals(errorMessage + "\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals(errorMessage + "\r\n", new String(bytes)); + } + + private class Service { @SuppressWarnings("unused") public String serviceMethod(byte[] bytes) { @@ -184,6 +228,13 @@ public class TcpInboundGatewayTests { } } + private class FailingService { + @SuppressWarnings("unused") + public String serviceMethod(byte[] bytes) { + throw new RuntimeException("Failed"); + } + } + private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { buff[i] = (byte) is.read(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index c15cf98040..74c3c82ffa 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -36,11 +36,13 @@ import java.util.concurrent.Executors; import javax.net.SocketFactory; import org.junit.Test; - import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory; import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory; @@ -585,4 +587,46 @@ public class TcpReceivingChannelAdapterTests { assertEquals("Test2", new ObjectInputStream(socket2.getInputStream()).readObject()); } + @Test + public void testException() throws Exception { + final int port = SocketTestUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + scf.setSerializer(serializer); + scf.setDeserializer(serializer); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + SubscribableChannel channel = new DirectChannel(); + adapter.setOutputChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService()); + channel.subscribe(handler); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test1\r\n".getBytes()); + socket.getOutputStream().write("Test2\r\n".getBytes()); + QueueChannel errorChannel = new QueueChannel(); + adapter.setErrorChannel(errorChannel); + Message message = errorChannel.receive(10000); + assertNotNull(message); + assertEquals("Failed", ((Exception) message.getPayload()).getCause().getMessage()); + message = errorChannel.receive(10000); + assertNotNull(message); + assertEquals("Failed", ((Exception) message.getPayload()).getCause().getMessage()); + } + + private class FailingService { + @SuppressWarnings("unused") + public String serviceMethod(byte[] bytes) { + throw new RuntimeException("Failed"); + } + } + + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index dc020c369d..3a164c7006 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -12,7 +12,10 @@ import org.apache.commons.logging.LogFactory; import org.junit.Ignore; import org.junit.Test; import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.support.MessageBuilder; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -130,5 +133,40 @@ public class UdpChannelAdapterTests { assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); } + @Test + public void testUnicastReceiverException() throws Exception { + SubscribableChannel channel = new DirectChannel(); + int port = SocketTestUtils.findAvailableUdpSocket(); + UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); +// SocketUtils.setLocalNicIfPossible(adapter); + adapter.setOutputChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService()); + channel.subscribe(handler); + QueueChannel errorChannel = new QueueChannel(); + adapter.setErrorChannel(errorChannel); + adapter.start(); + SocketTestUtils.waitListening(adapter); + + Message message = MessageBuilder.withPayload("ABCD".getBytes()).build(); + DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); + DatagramPacket packet = mapper.fromMessage(message); + packet.setSocketAddress(new InetSocketAddress("localhost", port)); + new DatagramSocket(SocketTestUtils.findAvailableUdpSocket()).send(packet); + Message receivedMessage = errorChannel.receive(2000); + assertNotNull(receivedMessage); + assertEquals("Failed", ((Exception) receivedMessage.getPayload()).getCause().getMessage()); + } + + private class FailingService { + @SuppressWarnings("unused") + public String serviceMethod(byte[] bytes) { + throw new RuntimeException("Failed"); + } + } + }