001/**
002 * Copyright 2005-2014 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 */
016package org.kuali.rice.ksb.messaging.exceptionhandling;
017
018import java.sql.Timestamp;
019
020import org.apache.commons.lang.StringUtils;
021import org.apache.log4j.Logger;
022import org.kuali.rice.core.api.config.property.ConfigContext;
023import org.kuali.rice.ksb.api.bus.ServiceConfiguration;
024import org.kuali.rice.ksb.messaging.PersistedMessageBO;
025import org.kuali.rice.ksb.service.KSBServiceLocator;
026import 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 */
034public 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        message = 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}