001 /**
002 * Copyright 2005-2012 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 package org.kuali.rice.ksb.messaging.exceptionhandling;
017
018 import java.sql.Timestamp;
019
020 import org.apache.commons.lang.StringUtils;
021 import org.apache.log4j.Logger;
022 import org.kuali.rice.core.api.config.property.ConfigContext;
023 import org.kuali.rice.ksb.api.bus.ServiceConfiguration;
024 import org.kuali.rice.ksb.messaging.PersistedMessageBO;
025 import org.kuali.rice.ksb.service.KSBServiceLocator;
026 import org.kuali.rice.ksb.util.KSBConstants;
027
028
029 /**
030 * Default implementation of the {@link MessageExceptionHandler} which handles exceptions thrown from message processing.
031 *
032 * @author Kuali Rice Team (rice.collab@kuali.org)
033 */
034 public class DefaultMessageExceptionHandler implements MessageExceptionHandler {
035
036 private static final Logger LOG = Logger.getLogger(DefaultMessageExceptionHandler.class);
037
038 private static final long DEFAULT_TIME_INCREMENT = 60 * 60 * 1000;
039
040 private static final int DEFAULT_MAX_RETRIES = 7;
041
042 public void handleException(Throwable throwable, PersistedMessageBO message, Object service) throws Exception {
043 if (isInException(message)) {
044 placeInException(throwable, message);
045 } else {
046 requeue(throwable, message);
047 }
048 }
049
050 public void handleExceptionLastDitchEffort(Throwable throwable, PersistedMessageBO message, Object service) throws Exception {
051 LOG.error("Complete failure when attempting to put message into exception routing! Message was: " + message, throwable);
052 }
053
054 public boolean isInException(PersistedMessageBO message) {
055 ServiceConfiguration serviceConfiguration = message.getMethodCall().getServiceConfiguration();
056
057 if (getImmediateExceptionRouting()) {
058 return true;
059 }
060
061 Integer globalMaxRetryAttempts = getGlobalMaxRetryAttempts();
062 if (globalMaxRetryAttempts != null) {
063 LOG.info("Global Max Retry has been set, so is overriding other max retry attempts.");
064 LOG.info("Global Max Retry count = " + globalMaxRetryAttempts + ".");
065 return (message.getRetryCount().intValue() >= globalMaxRetryAttempts.intValue());
066 }
067
068 if (serviceConfiguration.getRetryAttempts() > 0) {
069 LOG.info("Message set for retry exception handling. Message retry count = " + message.getRetryCount());
070 if (message.getRetryCount() >= serviceConfiguration.getRetryAttempts()) {
071 return true;
072 }
073 } else if (serviceConfiguration.getMillisToLive() > 0) {
074 LOG.info("Message set for time to live exception handling. Message expiration date = " + message.getExpirationDate().getTime());
075 if (System.currentTimeMillis() > message.getExpirationDate().getTime()) {
076 return true;
077 }
078 } else if (message.getRetryCount() >= this.getMaxRetryAttempts()) {
079 LOG.info("Message set for default exception handling. Comparing retry count = " + message.getRetryCount() + " against default max count.");
080 return true;
081 }
082 return false;
083 }
084
085 protected void requeue(Throwable throwable, PersistedMessageBO message) throws Exception {
086 Integer retryCount = message.getRetryCount();
087 message.setQueueStatus(KSBConstants.ROUTE_QUEUE_QUEUED);
088 long addMilliseconds = Math.round(getTimeIncrement() * Math.pow(2, retryCount));
089 Timestamp currentTime = message.getQueueDate();
090 Timestamp newTime = new Timestamp(currentTime.getTime() + addMilliseconds);
091 message.setQueueStatus(KSBConstants.ROUTE_QUEUE_QUEUED);
092 message.setRetryCount(new Integer(retryCount + 1));
093 message.setQueueDate(newTime);
094 scheduleExecution(throwable, message);
095 }
096
097 protected void placeInException(Throwable throwable, PersistedMessageBO message) throws Exception {
098 message.setQueueStatus(KSBConstants.ROUTE_QUEUE_EXCEPTION);
099 message.setQueueDate(new Timestamp(System.currentTimeMillis()));
100 KSBServiceLocator.getMessageQueueService().save(message);
101 }
102
103 protected void scheduleExecution(Throwable throwable, PersistedMessageBO message) throws Exception {
104 KSBServiceLocator.getExceptionRoutingService().scheduleExecution(throwable, message, null);
105 }
106
107 public Integer getMaxRetryAttempts() {
108 try {
109 return new Integer(ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.ROUTE_QUEUE_MAX_RETRY_ATTEMPTS_KEY));
110 } catch (NumberFormatException e) {
111 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");
112 return DEFAULT_MAX_RETRIES;
113 }
114 }
115
116 public Integer getGlobalMaxRetryAttempts() {
117 String globalMax = ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.ROUTE_QUEUE_MAX_RETRY_ATTEMPTS_OVERRIDE_KEY);
118 if (StringUtils.isBlank(globalMax)) {
119 return null;
120 }
121 try {
122 Integer globalMaxRetries = new Integer(globalMax);
123 if (globalMaxRetries >= 0) {
124 return globalMaxRetries;
125 }
126 } catch (NumberFormatException e) {
127 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");
128 }
129 return null;
130 }
131
132 public Long getTimeIncrement() {
133 try {
134 return new Long(ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.ROUTE_QUEUE_TIME_INCREMENT_KEY));
135 } catch (NumberFormatException e) {
136 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.");
137 return DEFAULT_TIME_INCREMENT;
138 }
139 }
140
141 public Boolean getImmediateExceptionRouting() {
142 return new Boolean(ConfigContext.getCurrentContextConfig().getProperty(KSBConstants.Config.IMMEDIATE_EXCEPTION_ROUTING));
143 }
144 }