From 30fd7decc8f8dc8da2f5607430763fcfb21e80a2 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 18 Aug 2008 18:40:39 +0000 Subject: [PATCH] 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. --- .../adapter/file/AbstractDirectorySource.java | 24 ++++++++++++------- .../adapter/ftp/FtpSourceTests.java | 15 ++++++------ .../endpoint/AbstractEndpoint.java | 2 +- .../endpoint/InboundChannelAdapter.java | 7 +++--- .../message/MessageDeliveryAware.java | 2 +- .../message/MessageExchangeTemplate.java | 18 +++++++++----- 6 files changed, 39 insertions(+), 29 deletions(-) diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java index 9442e341e5..2d88e10bdf 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java @@ -47,23 +47,25 @@ public abstract class AbstractDirectorySource implements PollableSource, M private final MessageCreator messageCreator; + public AbstractDirectorySource(MessageCreator messageCreator) { Assert.notNull(messageCreator, "The MessageCreator must not be null"); this.messageCreator = messageCreator; } + protected Backlog getDirectoryContentManager() { return this.directoryContentManager; } public MessageCreator getMessageCreator() { - return messageCreator; + return this.messageCreator; } @SuppressWarnings("unchecked") public final Message receive() { try { - refreshSnapshotAndMarkProcessing(directoryContentManager); + refreshSnapshotAndMarkProcessing(this.directoryContentManager); if (!(getDirectoryContentManager().getProcessingBuffer().isEmpty() & getDirectoryContentManager() .getBacklog().isEmpty())) { return buildNextMessage(); @@ -78,8 +80,9 @@ public abstract class AbstractDirectorySource implements PollableSource, 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 * directoryContentManager.fileProcessing(...) with the appropriate arguments. + * * @param directoryContentManager * @throws IOException */ @@ -91,9 +94,9 @@ public abstract class AbstractDirectorySource implements PollableSource, 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 implements PollableSource, M * @throws IOException */ protected Message 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 implements PollableSource, M /** * Returns the next file, based on the backlog data. + * * @return * @throws IOException */ diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java index 7f279b2cd7..73692557b5 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java +++ b/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> received = ftpSource.receive(); - ftpSource.onFailure(new MessagingException(received)); + ftpSource.onFailure(received, new Exception("test failure")); assertEquals(received, ftpSource.receive()); verify(globalMocks); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 95b8cd9b14..02e2da9de1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java index 0ba5faa18d..f573460d09 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java +++ b/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); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java index 21768e09a8..efd40b3049 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java +++ b/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); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java index 491a7065bd..fcb28f0536 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java +++ b/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); } }