Modified the 'onFailure' method signature in MessageDeliveryAware so that any Exception can be passed along with a separate Message parameter instead of always expecting a MessagingException.

org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java

@@ -47,23 +47,25 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 
 	private final MessageCreator<T, T> messageCreator;
 
+
 	public AbstractDirectorySource(MessageCreator<T, T> messageCreator) {
 		Assert.notNull(messageCreator, "The MessageCreator must not be null");
 		this.messageCreator = messageCreator;
 	}
 
+
 	protected Backlog<FileInfo> getDirectoryContentManager() {
 		return this.directoryContentManager;
 	}
 
 	public MessageCreator<T, T> getMessageCreator() {
-		return messageCreator;
+		return this.messageCreator;
 	}
 
 	@SuppressWarnings("unchecked")
 	public final Message<T> receive() {
 		try {
-			refreshSnapshotAndMarkProcessing(directoryContentManager);
+			refreshSnapshotAndMarkProcessing(this.directoryContentManager);
 			if (!(getDirectoryContentManager().getProcessingBuffer().isEmpty() & getDirectoryContentManager()
 					.getBacklog().isEmpty())) {
 				return buildNextMessage();
@@ -78,8 +80,9 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 	/**
 	 * Naive implementation that ignores thread safety. Subclasses that want to
 	 * be thread safe and use the reservation facilities of
-	 * {@link Backlog} override this method and call
+	 * {@link Backlog} should override this method and call
 	 * <code>directoryContentManager.fileProcessing(...)</code> with the appropriate arguments.
+	 * 
 	 * @param directoryContentManager
 	 * @throws IOException
 	 */
@@ -91,9 +94,9 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 
 	/**
 	 * Hook point for implementors to create the next message that should be
-	 * received. Implementations can use a File by File approach like
+	 * received. Implementations can use a File by File approach (like
 	 * FileSource). In cases where retrieval could be expensive because of
-	 * network latency a batched approach could be implemented here. See
+	 * network latency, a batched approach could be implemented here. See
 	 * FtpSource for an example.
 	 * 
 	 * @return the next message containing (part of) the unprocessed content of
@@ -101,20 +104,22 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 	 * @throws IOException
 	 */
 	protected Message<T> buildNextMessage() throws IOException {
-		return messageCreator.createMessage(retrieveNextPayload());
+		return this.messageCreator.createMessage(retrieveNextPayload());
 	}
 
 	public abstract void onSend(Message<?> message);
 
-	public void onFailure(MessagingException exception) {
+	public void onFailure(Message<?> failedMessage, Throwable exception) {
 		if (this.logger.isWarnEnabled()) {
-			logger.warn("Failure notification received by " + this.getClass().getSimpleName(), exception);
+			this.logger.warn("Failure notification received by [" + this.getClass().getSimpleName()
+					+ "] for message: " + failedMessage, exception);
 		}
-		directoryContentManager.processingFailed();
+		this.directoryContentManager.processingFailed();
 	}
 
 	/**
 	 * Constructs the snapshot by iterating files.
+	 * 
 	 * @param snapshot
 	 * @throws IOException
 	 */
@@ -122,6 +127,7 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 
 	/**
 	 * Returns the next file, based on the backlog data.
+	 * 
 	 * @return
 	 * @throws IOException
 	 */

org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java

@@ -37,11 +37,14 @@ import org.easymock.IAnswer;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;
+
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageCreator;
-import org.springframework.integration.message.MessagingException;
 
+/**
+ * @author Iwein Fuld
+ */
 @SuppressWarnings("unchecked")
 public class FtpSourceTests {
 
@@ -60,16 +63,11 @@ public class FtpSourceTests {
 
 	private Object[] globalMocks = new Object[] { messageCreator, ftpClient, ftpFile, ftpClientPool };
 
-	private static final String HOST = "testHost";
-
-	private static final String USER = "testUser";
-
-	private static final String PASS = "testPass";
-
 	private FtpSource ftpSource;
 
 	private Long size = 100l;
 
+
 	@Before
 	public void initializeFtpSource() {
 		ftpSource = new FtpSource(messageCreator, ftpClientPool);
@@ -80,6 +78,7 @@ public class FtpSourceTests {
 		reset(globalMocks);
 	}
 
+
 	@Test
 	public void retrieveSingleFile() throws Exception {
 
@@ -250,7 +249,7 @@ public class FtpSourceTests {
 				new GenericMessage(Arrays.asList(new File("test1")))).times(2);
 		replay(globalMocks);
 		Message<List<File>> received = ftpSource.receive();
-		ftpSource.onFailure(new MessagingException(received));
+		ftpSource.onFailure(received, new Exception("test failure"));
 		assertEquals(received, ftpSource.receive());
 		verify(globalMocks);
 	}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java

@@ -116,7 +116,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 			}
 			else {
 				this.handleException(new MessageHandlingException(message,
-						"failure occurred in endpoint's send operation", e));
+						"failure occurred in endpoint '" + this.toString() + "'", e));
 			}
 			return false;
 		}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java

@@ -41,12 +41,11 @@ public class InboundChannelAdapter extends AbstractEndpoint {
 			return sent;
 		}
 		catch (Exception e) {
-			MessagingException exception = (e instanceof MessagingException) ? (MessagingException) e
-					: new MessageDeliveryException(message, "channel-adapter failed to send message to target");
 			if (this.getSource() instanceof MessageDeliveryAware) {
-				((MessageDeliveryAware) this.getSource()).onFailure(exception);
+				((MessageDeliveryAware) this.getSource()).onFailure(message, e);
 			}
-			return false;
+			throw (e instanceof MessagingException) ? (MessagingException) e
+					: new MessageDeliveryException(message, "channel adapter failed to send message to target", e);
 		}
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java

@@ -32,6 +32,6 @@ public interface MessageDeliveryAware {
 	/**
 	 * Callback method invoked after a message delivery failure.
 	 */
-	void onFailure(MessagingException exception);
+	void onFailure(Message<?> failedMessage, Throwable t);
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java

@@ -226,8 +226,9 @@ public class MessageExchangeTemplate implements InitializingBean {
 	}
 
 	private boolean doReceiveAndForward(PollableSource<?> source, MessageTarget target) {
+		Message<?> message = null;
 		try {
-			Message<?> message = this.doReceive(source);
+			message = this.doReceive(source);
 			if (message == null) {
 				return false;
 			}
@@ -237,18 +238,23 @@ public class MessageExchangeTemplate implements InitializingBean {
 					((MessageDeliveryAware) source).onSend(message);
 				}
 				else {
-					((MessageDeliveryAware) source).onFailure(new MessageDeliveryException(message, "failed to send message"));
+					((MessageDeliveryAware) source).onFailure(message, new MessageDeliveryException(message, "failed to send message"));
 				}
 			}
 			return sent;
 		}
 		catch (Exception e) {
-			MessagingException exception = (e instanceof MessagingException) ? (MessagingException) e
-					: new MessagingException("exception occurred in receive-and-forward exchange", e);
 			if (source instanceof MessageDeliveryAware) {
-				((MessageDeliveryAware) source).onFailure(exception);
+				((MessageDeliveryAware) source).onFailure(message, e);
 			}
-			throw exception;
+			if (e instanceof MessagingException) {
+				throw (MessagingException) e;
+			}
+			String description = "exception occurred in receive-and-forward exchange";
+			if (message != null) {
+				throw new MessagingException(message, description, e);
+			}
+			throw new MessagingException(description, e);
 		}
 	}