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.

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<TcpConnectionSupport> connections = new LinkedList<TcpConnectionSupport>();
+	private final List<TcpConnectionSupport> connections = new LinkedList<TcpConnectionSupport>();
 
 	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());

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

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<byte[]>,
-		Deserializer<byte[]> {
+		Deserializer<byte[]>,
+		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);
+		}
+	}
+
 }

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

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

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

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;
+		}
 	}
 
 	/**

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

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

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<TcpDeserializationExceptionEvent> event =
+				new AtomicReference<TcpDeserializationExceptionEvent>();
+		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();
+	}
 
 }

src/reference/docbook/ip.xml

@@ -500,6 +500,14 @@
         </listitem>
       </itemizedlist>
     </para>
+    <para>
+        In addition, since <emphasis>version 4.0</emphasis> the standard deserializers discussed in
+        <xref linkend="connection-factories"/> now emit <classname>TcpDeserializationExceptionEvent</classname>s
+        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 <interfacename>ApplicationListener</interfacename>,
+        or see <xref linkend="applicationevent-inbound"/>, to capture these events, allowing analysis of the problem.
+    </para>
   </section>
   <section id="tcp-adapters">
     <title>TCP Adapters</title>

src/reference/docbook/whats-new.xml

@@ -341,5 +341,14 @@
 				See <xref linkend="jpa-retrieving-outbound-gateway"/> for more information.
 			</para>
 		</section>
+		<section id="4.0-tcp-deserializer-events">
+			<title>TCP Deserialization Events</title>
+			<para>
+				When one of the standard deserializers encounters a problem decoding the input stream to
+				a message, it will now emit a <classname>TcpDeserializationExceptionEvent</classname>, allowing
+				applications to examine the data at the point the exception occurred.
+				See <xref linkend="tcp-events"/> for more information.
+			</para>
+		</section>
 	</section>
 </chapter>