From 945bc9511fe09ac556f8c0a1cb5e11f3cc6e8227 Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Sat, 19 Jan 2008 21:00:44 +0000
Subject: [PATCH] ConcurrentHandler and DefaultMessageEndpoint now propagate
 rejection exceptions so that DispatcherTask has proper retry behavior, and
 PollingSchedule now implements equals and hashCode so that the handler map
 keys are recognized (handlers are added to the same task if they have the
 same schedule).

---
 .../endpoint/DefaultMessageEndpoint.java        |  4 ++++
 .../integration/handler/ConcurrentHandler.java  |  4 +---
 .../integration/scheduling/PollingSchedule.java | 17 +++++++++++++++++
 3 files changed, 22 insertions(+), 3 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
index facef65a6f..a59772beec 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
@@ -29,6 +29,7 @@ import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.MessageHandlerNotRunningException;
+import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.dispatcher.MessageSelectorRejectedException;
 import org.springframework.integration.handler.ConcurrentHandler;
 import org.springframework.integration.handler.MessageHandler;
@@ -216,6 +217,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 				replyChannel.send(replyMessage);
 			}
 		}
+		catch (MessageHandlerRejectedExecutionException e) {
+			throw e;
+		}
 		catch (Throwable t) {
 			if (this.errorHandler == null) {
 				throw new MessageHandlingException(
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java
index 952a8fa335..223d9102fe 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java
@@ -16,8 +16,6 @@
 
 package org.springframework.integration.handler;
 
-import java.util.concurrent.RejectedExecutionException;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -164,7 +162,7 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
 			this.executor.execute(new HandlerTask(message));
 			return null;
 		}
-		catch (RejectedExecutionException e) {
+		catch (RuntimeException e) {
 			throw new MessageHandlerRejectedExecutionException(e);
 		}
 	}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java
index 882764abd5..1263213e0e 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java
@@ -83,4 +83,21 @@ public class PollingSchedule implements Schedule {
 		this.fixedRate = fixedRate;
 	}
 
+	public int hashCode() {
+		return toString().hashCode() * 23;
+	}
+
+	public boolean equals(Object other) {
+		if (other == null) {
+			return false;
+		}
+		return (other instanceof PollingSchedule && 
+				this.toString().equals(((PollingSchedule) other).toString()));
+	}
+
+	public String toString() {
+		return "initialDelay=" + this.initialDelay + ", period=" + this.period +
+				", timeUnit=" + this.timeUnit + ", fixedRate=" + this.fixedRate;
+	}
+
 }