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

INT-1623 INT-1624 Add error-channel to ip inbound adapters and tcp inbound gateway - parsers, XSD and tests only - super classes already do the heavy lifting

Browse Source
This commit is contained in:
Gary Russell
2010-11-16 19:20:03 -05:00
parent 722c990cf4
commit bfeee039bb
9 changed files with 222 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractInboundGatewayParser.java
View File

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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java
View File

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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java
View File

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

27
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -48,6 +48,15 @@ its configuration specifies the number of threads.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -110,6 +119,15 @@ adapter.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -177,6 +195,15 @@ A connection factory is needed by an inbound adapter. The connection factory mus
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

17
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -29,6 +29,7 @@
so-timeout="32"
local-address="127.0.0.1"
task-executor="externalTE"
error-channel="errorChannel"
/>
<ip:udp-inbound-channel-adapter id="testInUdpMulticast"
@@ -52,6 +53,7 @@
<ip:tcp-inbound-channel-adapter id="testInTcp"
channel="tcpChannel"
error-channel="errorChannel"
connection-factory="cfS1"
/>
@@ -105,13 +107,26 @@
port="#{tcpIpUtils.findAvailableServerSocket(5800)}"
/>
<ip:tcp-inbound-gateway id="inGateway"
<ip:tcp-inbound-gateway id="inGateway1"
request-channel="tcpChannel"
reply-channel="replyChannel"
error-channel="errorChannel"
connection-factory="cfS2"
reply-timeout="456"
/>
<ip:tcp-connection-factory id="cfS3"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(5850)}"
/>
<ip:tcp-inbound-gateway id="inGateway2"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfS3"
reply-timeout="456"
/>
<ip:tcp-connection-factory id="cfC2"
type="client"
port="#{tcpIpUtils.findAvailableServerSocket(5900)}"

41
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.ip.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@@ -32,6 +33,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
@@ -85,8 +87,12 @@ public class ParserUnitTests {
TcpSendingMessageHandler tcpOut;
@Autowired
@Qualifier(value="inGateway")
TcpInboundGateway tcpInboundGateway;
@Qualifier(value="inGateway1")
TcpInboundGateway tcpInboundGateway1;
@Autowired
@Qualifier(value="inGateway2")
TcpInboundGateway tcpInboundGateway2;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpOutboundGateway#0")
@@ -134,6 +140,10 @@ public class ParserUnitTests {
@Qualifier(value="cfS2")
AbstractConnectionFactory cfS2;
@Autowired
@Qualifier(value="cfS3")
AbstractConnectionFactory cfS3;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#1")
TcpSendingMessageHandler tcpNewOut1;
@@ -150,6 +160,10 @@ public class ParserUnitTests {
@Qualifier(value="tcpNewIn2")
TcpReceivingChannelAdapter tcpNewIn2;
@Autowired
@Qualifier("errorChannel")
private MessageChannel errorChannel;
@Test
public void testInUdp() {
DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn);
@@ -163,6 +177,7 @@ public class ParserUnitTests {
assertEquals("ip:udp-inbound-channel-adapter", udpIn.getComponentType());
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertEquals(errorChannel, dfa.getPropertyValue("errorChannel"));
}
@Test
@@ -177,6 +192,7 @@ public class ParserUnitTests {
assertEquals(32, dfa.getPropertyValue("soTimeout"));
assertEquals("127.0.0.1", dfa.getPropertyValue("localAddress"));
assertNotSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
assertNull(dfa.getPropertyValue("errorChannel"));
}
@Test
@@ -185,6 +201,7 @@ public class ParserUnitTests {
assertSame(cfS1, dfa.getPropertyValue("serverConnectionFactory"));
assertEquals("testInTcp",tcpIn.getComponentName());
assertEquals("ip:tcp-inbound-channel-adapter", tcpIn.getComponentType());
assertEquals(errorChannel, dfa.getPropertyValue("errorChannel"));
}
@Test
@@ -239,12 +256,23 @@ public class ParserUnitTests {
}
@Test
public void testInGateway() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInboundGateway);
public void testInGateway1() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInboundGateway1);
assertSame(cfS2, dfa.getPropertyValue("connectionFactory"));
assertEquals(456L, dfa.getPropertyValue("replyTimeout"));
assertEquals("inGateway",tcpInboundGateway.getComponentName());
assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway.getComponentType());
assertEquals("inGateway1",tcpInboundGateway1.getComponentName());
assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway1.getComponentType());
assertEquals(errorChannel, dfa.getPropertyValue("errorChannel"));
}
@Test
public void testInGateway2() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInboundGateway2);
assertSame(cfS3, dfa.getPropertyValue("connectionFactory"));
assertEquals(456L, dfa.getPropertyValue("replyTimeout"));
assertEquals("inGateway2",tcpInboundGateway2.getComponentName());
assertEquals("ip:tcp-inbound-gateway", tcpInboundGateway2.getComponentType());
assertNull(dfa.getPropertyValue("errorChannel"));
}
@Test
@@ -353,6 +381,7 @@ public class ParserUnitTests {
public void testNewIn1() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn1);
assertSame(client1, dfa.getPropertyValue("clientConnectionFactory"));
assertNull(dfa.getPropertyValue("errorChannel"));
}
@Test

53
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java
View File

@@ -29,14 +29,19 @@ import java.util.Set;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageHandler;
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.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.channel.ChannelResolver;
public class TcpInboundGatewayTests {
@@ -177,6 +182,45 @@ public class TcpInboundGatewayTests {
assertTrue(results.remove("Echo:Test2\r\n"));
}
@Test
public void testErrorFlow() throws Exception {
final int port = SocketTestUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
scf.setSingleUse(true);
TcpInboundGateway gateway = new TcpInboundGateway();
gateway.setConnectionFactory(scf);
SubscribableChannel errorChannel = new DirectChannel();
final String errorMessage = "An error occurred";
errorChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel();
replyChannel.send(new GenericMessage<String>(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();

46
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

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

38
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java
View File

@@ -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<byte[]> 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");
}
}
}