001 /*
002 * Copyright 2006-2011 The Kuali Foundation
003 *
004 * Licensed under the Educational Community License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.opensource.org/licenses/ecl2.php
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017 package org.kuali.rice.ksb.messaging.exceptionhandling;
018
019 import java.sql.Timestamp;
020
021 import org.apache.commons.lang.StringUtils;
022 import org.apache.log4j.Logger;
023 import org.kuali.rice.core.api.config.property.ConfigContext;
024 import org.kuali.rice.ksb.api.bus.ServiceConfiguration;
025 import org.kuali.rice.ksb.messaging.PersistedMessageBO;
026 import org.kuali.rice.ksb.service.KSBServiceLocator;
027 import org.kuali.rice.ksb.util.KSBConstants;
028
029
030 /**
031 * Default implementation of the {@link MessageExceptionHandler} which handles exceptions thrown from message processing.
032 *
033 * @author Kuali Rice Team (rice.collab@kuali.org)
034 */
035 public class DefaultMessageExceptionHandler implements MessageExceptionHandler {
036
037 private static final Logger LOG = Logger.getLogger(DefaultMessageExceptionHandler.class);
038
039 private static final long DEFAULT_TIME_INCREMENT = 60 * 60 * 1000;
040
041 private static final int DEFAULT_MAX_RETRIES = 7;
042
043 public void handleException(Throwable throwable, PersistedMessageBO message, Object service) throws Exception {
044 if (isInException(message)) {
045 placeInException(throwable, message);
046 } else {
047 requeue(throwable, message);
048 }
049 }
050
051 public void handleExceptionLastDitchEffort(Throwable throwable, PersistedMessageBO message, Object service) throws Exception {
052 LOG.error("Complete failure when attempting to put message into exception routing! Message was: " + message, throwable);
053 }
054
055 public boolean isInException(PersistedMessageBO message) {
056 ServiceConfiguration serviceConfiguration = message.getMethodCall().getServiceConfiguration();
057
058 if (getImmediateExceptionRouting()) {
059 return true;
060 }
061
062 Integer globalMaxRetryAttempts = getGlobalMaxRetryAttempts();
063 if (globalMaxRetryAttempts != null) {
064 LOG.info("Global Max Retry has been set, so is overriding other max retry attempts.");
065 LOG.info("Global Max Retry count = " + globalMaxRetryAttempts + ".");
066 return (message.getRetryCount().intValue() >= globalMaxRetryAttempts.intValue());
067 }
068
069 if (serviceConfiguration.getRetryAttempts() > 0) {
070 LOG.info("Message set for retry exception handling. Message retry count = " + message.getRetryCount());
071 if (message.getRetryCount() >= serviceConfiguration.getRetryAttempts()) {
072 return true;
073 }
074 } else if (serviceConfiguration.getMillisToLive() > 0) {
075 LOG.info("Message set for time to live exception handling. Message expiration date = " + message.getExpirationDate().getTime());
076 if (System.currentTimeMillis() > message.getExpirationDate().getTime()) {
077 return true;
078 }
079 } else if (message.getRetryCount() >= this.getMaxRetryAttempts()) {
080 LOG.info("Message set for default exception handling. Comparing retry count = " + message.getRetryCount() + " against default max count.");
081 return true;
082 }
083 return false;
084 }
085
086 protected void requeue(Throwable throwable, PersistedMessageBO message) throws Exception {
087 Integer retryCount = message.getRetryCount();
088 message.setQueueStatus(KSBConstants.ROUTE_QUEUE_QUEUED);
089 long addMilliseconds = Math.round(getTimeIncrement() * Math.pow(2, retryCount));
090 Timestamp currentTime = message.getQueueDate();
091 Timestamp newTime = new Timestamp(currentTime.getTime() + addMilliseconds);
092 message.setQueueStatus(KSBConstants.ROUTE_QUEUE_QUEUED);
093 message.setRetryCount(new Integer(retryCount + 1));
094 message.setQueueDate(newTime);
095 scheduleExecution(throwable, message);
096 }
097
098 protected void placeInException(Throwable throwable, PersistedMessageBO message) throws Exception {
099 message.setQueueStatus(KSBConstants.ROUTE_QUEUE_EXCEPTION);
100 message.setQueueDate(new Timestamp(System.currentTimeMillis()));
101 KSBServiceLocator.getMessageQueueService().save(message);
102 }
103
104 protected void scheduleExecution(Throwable throwable, PersistedMessageBO message) throws Exception {
105 KSBServiceLocator.getExceptionRoutingService().scheduleExecution(throwable, message, null);
106 }
107
108 public Integer getMaxRetryAttempts() {
109 try {
110 return new Integer(ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.ROUTE_QUEUE_MAX_RETRY_ATTEMPTS_KEY));
111 } catch (NumberFormatException e) {
112 LOG.error("Constant '" + KSBConstants.Config.ROUTE_QUEUE_MAX_RETRY_ATTEMPTS_KEY + "' is not a number and is being " + "used as a default for exception messages. " + DEFAULT_MAX_RETRIES + " will be used as a retry limit until this number is fixed");
113 return DEFAULT_MAX_RETRIES;
114 }
115 }
116
117 public Integer getGlobalMaxRetryAttempts() {
118 String globalMax = ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.ROUTE_QUEUE_MAX_RETRY_ATTEMPTS_OVERRIDE_KEY);
119 if (StringUtils.isBlank(globalMax)) {
120 return null;
121 }
122 try {
123 Integer globalMaxRetries = new Integer(globalMax);
124 if (globalMaxRetries >= 0) {
125 return globalMaxRetries;
126 }
127 } catch (NumberFormatException e) {
128 LOG.error("Constant '" + KSBConstants.Config.ROUTE_QUEUE_MAX_RETRY_ATTEMPTS_OVERRIDE_KEY + "' is not a number and is being " + "used as a default for exception messages. " + DEFAULT_MAX_RETRIES + " will be used as a retry limit until this number is fixed");
129 }
130 return null;
131 }
132
133 public Long getTimeIncrement() {
134 try {
135 return new Long(ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.ROUTE_QUEUE_TIME_INCREMENT_KEY));
136 } catch (NumberFormatException e) {
137 LOG.error("Constant '" + KSBConstants.Config.ROUTE_QUEUE_TIME_INCREMENT_KEY + "' is not a number and will not be used " + "as the default time increment for exception routing. Default of " + DEFAULT_TIME_INCREMENT + " will be used.");
138 return DEFAULT_TIME_INCREMENT;
139 }
140 }
141
142 public Boolean getImmediateExceptionRouting() {
143 return new Boolean(ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.IMMEDIATE_EXCEPTION_ROUTING));
144 }
145 }