From 3e0f10a657a66e623b6fa073a9e18b6389a06772 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 28 Apr 2014 23:57:19 +0300 Subject: [PATCH] INT-1956 TCP: Emit Stream Decoding Events JIRA: https://jira.spring.io/browse/INT-1956 Emit an application event when a decoding exception occurs, allowing the user to examine the buffer at the time the exception occurred. INT-1956: Polishing Polishing Use `OP_READ` instead of `readyOps()` when removing interest. Polishing - Fix ConnectionTimeoutTests Test publisher was casting all events to TcpConnectionEvent. --- .../connection/AbstractConnectionFactory.java | 11 ++- .../TcpDeserializationExceptionEvent.java | 51 ++++++++++ .../AbstractByteArraySerializer.java | 26 +++++- .../serializer/ByteArrayCrLfSerializer.java | 56 ++++++----- .../ByteArrayLengthHeaderSerializer.java | 80 ++++++++++------ .../serializer/ByteArrayRawSerializer.java | 47 ++++++---- .../ByteArraySingleTerminatorSerializer.java | 50 ++++++---- .../serializer/ByteArrayStxEtxSerializer.java | 53 +++++++---- .../connection/ConnectionTimeoutTests.java | 5 +- .../tcp/serializer/DeserializationTests.java | 92 +++++++++++++++---- src/reference/docbook/ip.xml | 8 ++ src/reference/docbook/whats-new.xml | 9 ++ 12 files changed, 361 insertions(+), 127 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpDeserializationExceptionEvent.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 5859385d50..3ac7ffa90c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -89,6 +89,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport private volatile Deserializer deserializer = new ByteArrayCrLfSerializer(); + private volatile boolean deserializerSet; + private volatile Serializer serializer = new ByteArrayCrLfSerializer(); private volatile TcpMessageMapper mapper = new TcpMessageMapper(); @@ -103,7 +105,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport private volatile boolean lookupHost = true; - private volatile List connections = new LinkedList(); + private final List connections = new LinkedList(); private volatile TcpSocketSupport tcpSocketSupport = new DefaultTcpSocketSupport(); @@ -130,6 +132,10 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; + if (!this.deserializerSet && this.deserializer instanceof ApplicationEventPublisherAware) { + ((ApplicationEventPublisherAware) this.deserializer) + .setApplicationEventPublisher(applicationEventPublisher); + } } protected ApplicationEventPublisher getApplicationEventPublisher() { @@ -345,6 +351,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport */ public void setDeserializer(Deserializer deserializer) { this.deserializer = deserializer; + this.deserializerSet = true; } /** @@ -585,7 +592,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport logger.debug("Selection key no longer valid"); } else if (key.isReadable()) { - key.interestOps(key.interestOps() - key.readyOps()); + key.interestOps(key.interestOps() - SelectionKey.OP_READ); final TcpNioConnection connection; connection = (TcpNioConnection) key.attachment(); connection.setLastRead(System.currentTimeMillis()); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpDeserializationExceptionEvent.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpDeserializationExceptionEvent.java new file mode 100644 index 0000000000..7e2d23a519 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpDeserializationExceptionEvent.java @@ -0,0 +1,51 @@ +/* + * Copyright 2014 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp.connection; + +import org.springframework.integration.ip.event.IpIntegrationEvent; + +/** + * Event representing an exception while decoding an incoming stream. + * Contains the buffer of data decoded so far and the offset in the + * buffer where the exception occurred, if available, otherwise -1. + * + * @author Gary Russell + * @since 4.0 + * + */ +public class TcpDeserializationExceptionEvent extends IpIntegrationEvent { + + private static final long serialVersionUID = 8812537718016054732L; + + private final byte[] buffer; + + private final int offset; + + public TcpDeserializationExceptionEvent(Object source, Throwable cause, byte[] buffer, int offset) { + super(source, cause); + this.buffer = buffer; + this.offset = offset; + } + + public byte[] getBuffer() { + return buffer; + } + + public int getOffset() { + return offset; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java index debe05f81c..65d8ad7428 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -20,8 +20,12 @@ import java.io.IOException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; +import org.springframework.integration.ip.tcp.connection.TcpDeserializationExceptionEvent; /** * Base class for (de)serializers that provide a mechanism to @@ -33,12 +37,15 @@ import org.springframework.core.serializer.Serializer; */ public abstract class AbstractByteArraySerializer implements Serializer, - Deserializer { + Deserializer, + ApplicationEventPublisherAware { protected int maxMessageSize = 2048; protected final Log logger = LogFactory.getLog(this.getClass()); + private ApplicationEventPublisher applicationEventPublisher; + /** * The maximum supported message size for this serializer. * Default 2048. @@ -57,6 +64,11 @@ public abstract class AbstractByteArraySerializer implements this.maxMessageSize = maxMessageSize; } + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + protected void checkClosure(int bite) throws IOException { if (bite < 0) { logger.debug("Socket closed during message assembly"); @@ -80,4 +92,14 @@ public abstract class AbstractByteArraySerializer implements return assembledData; } + protected void publishEvent(Exception cause, byte[] buffer, int offset) { + TcpDeserializationExceptionEvent event = new TcpDeserializationExceptionEvent(this, cause, buffer, offset); + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(event); + } + else if (logger.isTraceEnabled()) { + logger.trace("No event publisher for " + event); + } + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java index 4bdc94fd92..610492edd5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -33,47 +33,59 @@ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer { private static final byte[] CRLF = "\r\n".getBytes(); /** - * Reads the data in the inputstream to a byte[]. Data must be terminated + * Reads the data in the inputStream to a byte[]. Data must be terminated * by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream * is closed immediately after the \r\n (i.e. no data is in the process of * being read). */ + @Override public byte[] deserialize(InputStream inputStream) throws IOException { byte[] buffer = new byte[this.maxMessageSize]; int n = this.fillToCrLf(inputStream, buffer); - byte[] assembledData = this.copyToSizedArray(buffer, n); - return assembledData; + return this.copyToSizedArray(buffer, n); } - public int fillToCrLf(InputStream inputStream, byte[] buffer) - throws IOException, SoftEndOfStreamException { + public int fillToCrLf(InputStream inputStream, byte[] buffer) throws IOException { int n = 0; int bite; if (logger.isDebugEnabled()) { logger.debug("Available to read:" + inputStream.available()); } - while (true) { - bite = inputStream.read(); -// logger.debug("Read:" + (char) bite); - if (bite < 0 && n == 0) { - throw new SoftEndOfStreamException("Stream closed between payloads"); + try { + while (true) { + bite = inputStream.read(); + if (bite < 0 && n == 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + checkClosure(bite); + if (n > 0 && bite == '\n' && buffer[n-1] == '\r') { + break; + } + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("CRLF not found before max message length: " + + this.maxMessageSize); + } } - checkClosure(bite); - if (n > 0 && bite == '\n' && buffer[n-1] == '\r') { - break; - } - buffer[n++] = (byte) bite; - if (n >= this.maxMessageSize) { - throw new IOException("CRLF not found before max message length: " - + this.maxMessageSize); - } - }; - return n-1; // trim \r + return n-1; // trim \r + } + catch (SoftEndOfStreamException e) { + throw e; + } + catch (IOException e) { + publishEvent(e, buffer, n); + throw e; + } + catch (RuntimeException e) { + publishEvent(e, buffer, n); + throw e; + } } /** * Writes the byte[] to the stream and appends \r\n. */ + @Override public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { outputStream.write(bytes); outputStream.write(CRLF); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java index 84d3eb673e..a44372ffb6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java @@ -102,13 +102,24 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer if (logger.isDebugEnabled()) { logger.debug("Message length is " + messageLength); } - if (messageLength > this.maxMessageSize) { - throw new IOException("Message length " + messageLength + - " exceeds max message length: " + this.maxMessageSize); + byte[] messagePart = null; + try { + if (messageLength > this.maxMessageSize) { + throw new IOException("Message length " + messageLength + + " exceeds max message length: " + this.maxMessageSize); + } + messagePart = new byte[messageLength]; + read(inputStream, messagePart, false); + return messagePart; + } + catch (IOException e) { + publishEvent(e, messagePart, -1); + throw e; + } + catch (RuntimeException e) { + publishEvent(e, messagePart, -1); + throw e; } - byte[] messagePart = new byte[messageLength]; - read(inputStream, messagePart, false); - return messagePart; } /** @@ -204,29 +215,42 @@ public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer */ protected int readHeader(InputStream inputStream) throws IOException { byte[] lengthPart = new byte[this.headerSize]; - int status = read(inputStream, lengthPart, true); - if (status < 0) { - throw new SoftEndOfStreamException("Stream closed between payloads"); - } - int messageLength; - switch (this.headerSize) { - case HEADER_SIZE_INT: - messageLength = ByteBuffer.wrap(lengthPart).getInt(); - if (messageLength < 0) { - throw new IllegalArgumentException("Length header:" - + messageLength - + " is negative"); + try { + int status = read(inputStream, lengthPart, true); + if (status < 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); } - break; - case HEADER_SIZE_UNSIGNED_BYTE: - messageLength = ByteBuffer.wrap(lengthPart).get() & 0xff; - break; - case HEADER_SIZE_UNSIGNED_SHORT: - messageLength = ByteBuffer.wrap(lengthPart).getShort() & 0xffff; - break; - default: - throw new IllegalArgumentException("Bad header size:" + headerSize); + int messageLength; + switch (this.headerSize) { + case HEADER_SIZE_INT: + messageLength = ByteBuffer.wrap(lengthPart).getInt(); + if (messageLength < 0) { + throw new IllegalArgumentException("Length header:" + + messageLength + + " is negative"); + } + break; + case HEADER_SIZE_UNSIGNED_BYTE: + messageLength = ByteBuffer.wrap(lengthPart).get() & 0xff; + break; + case HEADER_SIZE_UNSIGNED_SHORT: + messageLength = ByteBuffer.wrap(lengthPart).getShort() & 0xffff; + break; + default: + throw new IllegalArgumentException("Bad header size:" + headerSize); + } + return messageLength; + } + catch (SoftEndOfStreamException e) { + throw e; + } + catch (IOException e) { + publishEvent(e, lengthPart, -1); + throw e; + } + catch (RuntimeException e) { + publishEvent(e, lengthPart, -1); + throw e; } - return messageLength; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java index 82e6bce7bf..e1b1718273 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -35,12 +35,14 @@ import java.io.OutputStream; */ public class ByteArrayRawSerializer extends AbstractByteArraySerializer { + @Override public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { outputStream.write(bytes); outputStream.flush(); } + @Override public byte[] deserialize(InputStream inputStream) throws IOException { byte[] buffer = new byte[this.maxMessageSize]; int n = 0; @@ -48,23 +50,36 @@ public class ByteArrayRawSerializer extends AbstractByteArraySerializer { if (logger.isDebugEnabled()) { logger.debug("Available to read:" + inputStream.available()); } - while (bite >= 0) { - bite = inputStream.read(); - if (bite < 0) { - if (n == 0) { - throw new SoftEndOfStreamException("Stream closed between payloads"); + try { + while (bite >= 0) { + bite = inputStream.read(); + if (bite < 0) { + if (n == 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + break; + } + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("Socket was not closed before max message length: " + + this.maxMessageSize); } - break; } - buffer[n++] = (byte) bite; - if (n >= this.maxMessageSize) { - throw new IOException("Socket was not closed before max message length: " - + this.maxMessageSize); - } - }; - byte[] assembledData = new byte[n]; - System.arraycopy(buffer, 0, assembledData, 0, n); - return assembledData; + byte[] assembledData = new byte[n]; + System.arraycopy(buffer, 0, assembledData, 0, n); + return assembledData; + } + catch (SoftEndOfStreamException e) { + throw e; + } + catch (IOException e) { + publishEvent(e, buffer, n); + throw e; + } + catch (RuntimeException e) { + publishEvent(e, buffer, n); + throw e; + } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java index 39bee41df3..c26e04c3a0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java @@ -37,7 +37,7 @@ public class ByteArraySingleTerminatorSerializer extends AbstractByteArraySerial } /** - * Reads the data in the inputstream to a byte[]. Data must be terminated + * Reads the data in the inputStream to a byte[]. Data must be terminated * by a single byte. Throws a {@link SoftEndOfStreamException} if the stream * is closed immediately after the terminator (i.e. no data is in the process of * being read). @@ -50,24 +50,38 @@ public class ByteArraySingleTerminatorSerializer extends AbstractByteArraySerial if (logger.isDebugEnabled()) { logger.debug("Available to read:" + inputStream.available()); } - while (true) { - bite = inputStream.read(); - if (bite < 0 && n == 0) { - throw new SoftEndOfStreamException("Stream closed between payloads"); + try { + while (true) { + bite = inputStream.read(); + if (bite < 0 && n == 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + checkClosure(bite); + if (bite == terminator) { + break; + } + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("Terminator '0x" + Integer.toHexString(terminator & 0xff) + + "' not found before max message length: " + + this.maxMessageSize); + } } - checkClosure(bite); - if (bite == terminator) { - break; - } - buffer[n++] = (byte) bite; - if (n >= this.maxMessageSize) { - throw new IOException("LF not found before max message length: " - + this.maxMessageSize); - } - }; - byte[] assembledData = new byte[n]; - System.arraycopy(buffer, 0, assembledData, 0, n); - return assembledData; + byte[] assembledData = new byte[n]; + System.arraycopy(buffer, 0, assembledData, 0, n); + return assembledData; + } + catch (SoftEndOfStreamException e) { + throw e; + } + catch (IOException e) { + publishEvent(e, buffer, n); + throw e; + } + catch (RuntimeException e) { + publishEvent(e, buffer, n); + throw e; + } } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java index 429a3936ad..7e168f1503 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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. @@ -24,53 +24,66 @@ import org.springframework.integration.mapping.MessageMappingException; /** * Reads data in an InputStream to a byte[]; data must be prefixed by <stx> and - * terminated by <etx> (not included in resulting byte[]). + * terminated by <etx> (not included in resulting byte[]). * Writes a byte[] to an OutputStream prefixed by <stx> terminated by <etx> - * + * * @author Gary Russell * @since 2.0 */ public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer { public static final int STX = 0x02; - + public static final int ETX = 0x03; /** - * Reads the data in the inputstream to a byte[]. Data must be prefixed + * Reads the data in the inputStream to a byte[]. Data must be prefixed * with an ASCII STX character, and terminated with an ASCII ETX character. * Throws a {@link SoftEndOfStreamException} if the stream * is closed immediately before the STX (i.e. no data is in the process of - * being read). - * + * being read). + * */ + @Override public byte[] deserialize(InputStream inputStream) throws IOException { int bite = inputStream.read(); if (bite < 0) { throw new SoftEndOfStreamException("Stream closed between payloads"); } - if (bite != STX) { - throw new MessageMappingException("Expected STX to begin message"); - } - byte[] buffer = new byte[this.maxMessageSize]; + byte[] buffer = null; int n = 0; - while ((bite = inputStream.read()) != ETX) { - checkClosure(bite); - buffer[n++] = (byte) bite; - if (n >= this.maxMessageSize) { - throw new IOException("ETX not found before max message length: " - + this.maxMessageSize); + try { + if (bite != STX) { + throw new MessageMappingException("Expected STX to begin message"); } + buffer = new byte[this.maxMessageSize]; + while ((bite = inputStream.read()) != ETX) { + checkClosure(bite); + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("ETX not found before max message length: " + + this.maxMessageSize); + } + } + byte[] assembledData = new byte[n]; + System.arraycopy(buffer, 0, assembledData, 0, n); + return assembledData; + } + catch (IOException e) { + publishEvent(e, buffer, n); + throw e; + } + catch (RuntimeException e) { + publishEvent(e, buffer, n); + throw e; } - byte[] assembledData = new byte[n]; - System.arraycopy(buffer, 0, assembledData, 0, n); - return assembledData; } /** * Writes the byte[] to the stream, prefixed by an ASCII STX character and * terminated with an ASCII ETX character. */ + @Override public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { outputStream.write(STX); outputStream.write(bytes); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java index c6e19dbe70..828a396ecb 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionTimeoutTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -271,8 +271,7 @@ public class ConnectionTimeoutTests { client.setApplicationEventPublisher(new ApplicationEventPublisher() { @Override public void publishEvent(ApplicationEvent event) { - TcpConnectionEvent tcpEvent = (TcpConnectionEvent) event; - if (tcpEvent instanceof TcpConnectionCloseEvent) { + if (event instanceof TcpConnectionCloseEvent) { clientClosedLatch.countDown(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java index a713cc9534..7e22122ca6 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java @@ -16,7 +16,11 @@ package org.springframework.integration.ip.tcp.serializer; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; import java.io.ByteArrayInputStream; @@ -24,12 +28,16 @@ import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; import javax.net.ServerSocketFactory; import org.junit.Test; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.DefaultDeserializer; +import org.springframework.integration.ip.tcp.connection.TcpDeserializationExceptionEvent; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.test.util.SocketUtils; @@ -242,22 +250,74 @@ public class DeserializationTests { latch.countDown(); } - @Test - public void canDeserializeMultipleSubsequentTerminators() throws IOException { - byte terminator = (byte) '\n'; - ByteArraySingleTerminatorSerializer serializer = new ByteArraySingleTerminatorSerializer(terminator); - ByteArrayInputStream inputStream = new ByteArrayInputStream("s\n\n".getBytes()); + @Test + public void canDeserializeMultipleSubsequentTerminators() throws IOException { + byte terminator = (byte) '\n'; + ByteArraySingleTerminatorSerializer serializer = new ByteArraySingleTerminatorSerializer(terminator); + ByteArrayInputStream inputStream = new ByteArrayInputStream("s\n\n".getBytes()); - try { - byte[] bytes = serializer.deserialize(inputStream); - assertEquals(1, bytes.length); - assertEquals("s".getBytes()[0], bytes[0]); - bytes = serializer.deserialize(inputStream); - assertEquals(0, bytes.length); - } - finally { - inputStream.close(); - } - } + try { + byte[] bytes = serializer.deserialize(inputStream); + assertEquals(1, bytes.length); + assertEquals("s".getBytes()[0], bytes[0]); + bytes = serializer.deserialize(inputStream); + assertEquals(0, bytes.length); + } + finally { + inputStream.close(); + } + } + + @Test + public void deserializationEvents() throws Exception { + doDeserialize(new ByteArrayCrLfSerializer(), "CRLF not found before max message length: 5"); + doDeserialize(new ByteArrayLengthHeaderSerializer(), "Message length 1718579042 exceeds max message length: 5"); + TcpDeserializationExceptionEvent event = doDeserialize(new ByteArrayLengthHeaderSerializer(), + "Stream closed after 3 of 4", new byte[] { 0, 0, 0 }, 5); // closed during header read + assertEquals(-1, event.getOffset()); + assertEquals(new String(new byte[] { 0, 0, 0 }), new String(event.getBuffer()).substring(0, 3)); + event = doDeserialize(new ByteArrayLengthHeaderSerializer(), + "Stream closed after 1 of 2", new byte[] { 0, 0, 0, 2, 7 }, 5); // closed during data read + assertEquals(-1, event.getOffset()); + assertEquals(new String(new byte[] { 7 }), new String(event.getBuffer()).substring(0, 1)); + doDeserialize(new ByteArrayLfSerializer(), "Terminator '0xa' not found before max message length: 5"); + doDeserialize(new ByteArrayRawSerializer(), "Socket was not closed before max message length: 5"); + doDeserialize(new ByteArraySingleTerminatorSerializer((byte) 0xfe), "Terminator '0xfe' not found before max message length: 5"); + doDeserialize(new ByteArrayStxEtxSerializer(), "Expected STX to begin message"); + event = doDeserialize(new ByteArrayStxEtxSerializer(), + "Socket closed during message assembly", new byte[] { 0x02, 0, 0 }, 5); + assertEquals(2, event.getOffset()); + } + + private TcpDeserializationExceptionEvent doDeserialize(AbstractByteArraySerializer deser, String expectedMessage) { + return doDeserialize(deser, expectedMessage, "foobar".getBytes(), 5); + } + + private TcpDeserializationExceptionEvent doDeserialize(AbstractByteArraySerializer deser, String expectedMessage, + byte[] data, int mms) { + final AtomicReference event = + new AtomicReference(); + class Publisher implements ApplicationEventPublisher { + + @Override + public void publishEvent(ApplicationEvent anEvent) { + event.set((TcpDeserializationExceptionEvent) anEvent); + } + } + Publisher publisher = new Publisher(); + ByteArrayInputStream bais = new ByteArrayInputStream(data); + deser.setApplicationEventPublisher(publisher); + deser.setMaxMessageSize(mms); + try { + deser.deserialize(bais); + fail("expected exception"); + } + catch (Exception e) { + assertNotNull(event.get()); + assertSame(e, event.get().getCause()); + assertThat(e.getMessage(), containsString(expectedMessage)); + } + return event.get(); + } } diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml index 3656b87973..e7c233eaea 100644 --- a/src/reference/docbook/ip.xml +++ b/src/reference/docbook/ip.xml @@ -500,6 +500,14 @@ + + In addition, since version 4.0 the standard deserializers discussed in + now emit TcpDeserializationExceptionEvents + when problems are encountered decoding the data stream. These events contain the exception, the + buffer that was in the process of being built, and an offset into the buffer (if available) at the + point the exception occurred. Applications can use a normal ApplicationListener, + or see , to capture these events, allowing analysis of the problem. +
TCP Adapters diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 16bcea0b0f..349ace4e79 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -341,5 +341,14 @@ See for more information.
+
+ TCP Deserialization Events + + When one of the standard deserializers encounters a problem decoding the input stream to + a message, it will now emit a TcpDeserializationExceptionEvent, allowing + applications to examine the data at the point the exception occurred. + See for more information. + +