/*
 *  Copyright 2017 Anyware Services
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package org.ametys.plugins.messagingconnector;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.apache.avalon.framework.activity.Initializable;
import org.apache.avalon.framework.service.ServiceException;
import org.apache.avalon.framework.service.ServiceManager;
import org.apache.avalon.framework.service.Serviceable;
import org.apache.commons.lang3.StringUtils;

import org.ametys.core.user.CurrentUserProvider;
import org.ametys.core.user.UserIdentity;
import org.ametys.core.user.population.UserPopulationDAO;
import org.ametys.core.userpref.UserPreferencesException;
import org.ametys.core.userpref.UserPreferencesManager;
import org.ametys.plugins.explorer.calendars.EventRecurrenceTypeEnum;
import org.ametys.plugins.messagingconnector.MessagingConnectorException.ExceptionType;
import org.ametys.runtime.config.Config;
import org.ametys.runtime.plugin.component.AbstractLogEnabled;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

/**
 * Abstract implementation of {@link MessagingConnector} with cache.
 *
 */
public abstract class AbstractMessagingConnector extends AbstractLogEnabled implements MessagingConnector, Initializable, Serviceable
{
    /** Duration of the cache for timeout errors, no calls will be done again for one user */
    private static final int TIMEOUT_CACHE_DURATION_SECONDS = 30;
    
    /** The user population DAO */
    protected UserPopulationDAO _userPopulationDAO;
    /** The user preferences */
    protected UserPreferencesManager _userPref;
    /** The crypto helper */
    protected CryptoHelper _cryptoHelper;
    /** The current user provider */
    protected CurrentUserProvider _currentUserProvider;

    private Cache<EventCacheKey, List<CalendarEvent>> _eventsCache;
    private Cache<EventCountCacheKey, Integer> _eventsCountCache;
    private Cache<EmailCacheKey, List<EmailMessage>> _emailsCache;
    private Cache<UserIdentity, Integer> _emailsCountCache;
    // Error caches to avoid calling the server again too quickly
    
    private Cache<MessagingConnectorException.ExceptionType, Set<UserIdentity>> _errorCache;
    private Cache<MessagingConnectorException.ExceptionType, Set<UserIdentity>> _timeoutErrorCache;
    
    private List<String> _populationIds;

    public void service(ServiceManager manager) throws ServiceException
    {
        _userPopulationDAO = (UserPopulationDAO) manager.lookup(UserPopulationDAO.ROLE);
        _userPref = (UserPreferencesManager) manager.lookup(UserPreferencesManager.ROLE);
        _cryptoHelper = (CryptoHelper) manager.lookup(CryptoHelper.ROLE);
        _currentUserProvider = (CurrentUserProvider) manager.lookup(CurrentUserProvider.ROLE);
    }
    
    @Override
    public void initialize()
    {
        _populationIds = new ArrayList<>();
        
        String populationIdsAsString = Config.getInstance().getValueAsString("org.ametys.plugins.messagingconnector.population");
        if (StringUtils.isNotBlank(populationIdsAsString))
        {
            List<String> userPopulationsIds = _userPopulationDAO.getUserPopulationsIds();
            String[] populationIds = StringUtils.split(populationIdsAsString, ",");
            
            List<String> wrongPopulationIds = new ArrayList<>();
            for (String populationId : populationIds)
            {
                String populationIdTrimed = StringUtils.trim(populationId);
                if (!userPopulationsIds.contains(populationIdTrimed))
                {
                    wrongPopulationIds.add(populationIdTrimed);
                }
                else
                {
                    _populationIds.add(populationIdTrimed);
                }
            }
            
            if (!wrongPopulationIds.isEmpty())
            {
                throw new IllegalStateException("The following population ids defined in the configuration parameter 'population id' for the messaging connector do not exist : " + wrongPopulationIds);
            }
        }
        
        Long maxCacheSizeConf = Config.getInstance().getValueAsLong("org.ametys.plugins.messagingconnector.cache.maxsize");
        Long maxCacheSize = (long) (maxCacheSizeConf != null ? maxCacheSizeConf.intValue() : 1000);
        
        Long cacheTtlConf = Config.getInstance().getValueAsLong("org.ametys.plugins.messagingconnector.cache.ttl");
        Long cacheTtl = (long) (cacheTtlConf != null && cacheTtlConf.intValue() > 0 ? cacheTtlConf.intValue() : 60);

        CacheBuilder<Object, Object> cacheBuilder = CacheBuilder.newBuilder().expireAfterWrite(cacheTtl, TimeUnit.MINUTES);
        CacheBuilder<Object, Object> timeoutCacheBuilder = CacheBuilder.newBuilder().expireAfterWrite(TIMEOUT_CACHE_DURATION_SECONDS, TimeUnit.SECONDS);
        
        if (maxCacheSize > 0)
        {
            cacheBuilder.maximumSize(maxCacheSize);
            timeoutCacheBuilder.maximumSize(maxCacheSize);
        }
        
        _eventsCache = cacheBuilder.<EventCacheKey, List<CalendarEvent>>build();
        _eventsCountCache = cacheBuilder.<EventCountCacheKey, Integer>build(); 
        _emailsCache = cacheBuilder.<EmailCacheKey, List<EmailMessage>>build();
        _emailsCountCache = cacheBuilder.<UserIdentity, Integer>build();
        
        _errorCache = cacheBuilder.<MessagingConnectorException.ExceptionType, Set<UserIdentity>>build();
        _timeoutErrorCache = timeoutCacheBuilder.<MessagingConnectorException.ExceptionType, Set<UserIdentity>>build();
    }
    
    @Override
    public List<String> getAllowedPopulationIds()
    {
        if (_populationIds.isEmpty())
        {
            List<String> userPopulationsIds = _userPopulationDAO.getUserPopulationsIds();
            if (userPopulationsIds.size() == 1)
            {
                return userPopulationsIds;
            }
            
            throw new IllegalStateException("There is more than one population defined. You must set the configuration parameter 'population id' for the messaging connector");
        }
        else
        {
            return _populationIds;
        }
    }
    
    /**
     * True if the user is allowed
     * @param userIdentity the user identity
     * @return true if the user is allowed
     */
    protected boolean isAllowed(UserIdentity userIdentity)
    {
        if (userIdentity == null)
        {
            getLogger().warn("There is no connected user to get user's mails or events from messaging connector");
            return false;
        }
        
        List<String> allowedPopulations = getAllowedPopulationIds();
        if (!allowedPopulations.contains(userIdentity.getPopulationId()))
        {
            getLogger().warn("The user " + userIdentity + " does not belong to any authorized user populations for messaging connector " + allowedPopulations);
            return false;
        }
        
        return true;
    }
    
    @Override
    public List<CalendarEvent> getEvents(UserIdentity userIdentity, int maxDays, int maxEvents) throws MessagingConnectorException
    {
        if (!isAllowed(userIdentity))
        {
            return new ArrayList<>();
        }

        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(userIdentity);
        
        try
        {
            EventCacheKey eventCacheKey = new EventCacheKey(userIdentity, maxDays, maxEvents);
            List<CalendarEvent> events = _eventsCache.getIfPresent(eventCacheKey);
            if (events == null)
            {
                events = internalGetEvents(userIdentity, maxDays, maxEvents);
                _eventsCache.put(eventCacheKey, events);
            }
            return events;
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(userIdentity, e.getType());
            throw e;
        }
    }
    
    @Override
    public int getEventsCount(UserIdentity userIdentity, int maxDays) throws MessagingConnectorException
    {
        if (!isAllowed(userIdentity))
        {
            return 0;
        }
        
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(userIdentity);
        
        try
        {
            EventCountCacheKey eventCountCacheKey = new EventCountCacheKey(userIdentity, maxDays);
            Integer eventsCount = _eventsCountCache.getIfPresent(eventCountCacheKey);
            if (eventsCount == null)
            {
                eventsCount = internalGetEventsCount(userIdentity, maxDays);
                _eventsCountCache.put(eventCountCacheKey, eventsCount);
            }
            return eventsCount;
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(userIdentity, e.getType());
            throw e;
        }
    }

    @Override
    public List<EmailMessage> getUnreadEmails(UserIdentity userIdentity, int maxEmails) throws MessagingConnectorException
    {
        if (!isAllowed(userIdentity))
        {
            return new ArrayList<>();
        }
        
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(userIdentity);
        
        try
        {
            EmailCacheKey emailCacheKey = new EmailCacheKey(userIdentity, maxEmails);
            List<EmailMessage> emails = _emailsCache.getIfPresent(emailCacheKey);
            if (emails == null)
            {
                emails = internalGetEmails(userIdentity, maxEmails);
                _emailsCache.put(emailCacheKey, emails);
            }
            return emails;
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(userIdentity, e.getType());
            throw e;
        }
    }

    @Override
    public int getUnreadEmailCount(UserIdentity userIdentity) throws MessagingConnectorException
    {
        if (!isAllowed(userIdentity))
        {
            return 0;
        }
        
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(userIdentity);
        
        try
        {
            Integer emailsCount = _emailsCountCache.getIfPresent(userIdentity);
            if (emailsCount == null)
            {
                emailsCount = internalGetEmailsCount(userIdentity);
                _emailsCountCache.put(userIdentity, emailsCount);
            }
            return emailsCount;
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(userIdentity, e.getType());
            throw e;
        }
    }
    
    /**
     * Get upcoming events (no caching)
     * @param userIdentity The user identity
     * @param maxDays The maximum number of days to search for
     * @param maxEvents The maximum number of events to retrieve
     * @return The calendar events
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected abstract List<CalendarEvent> internalGetEvents(UserIdentity userIdentity, int maxDays, int maxEvents) throws MessagingConnectorException;
    
    /**
     * Get upcoming events count (no caching)
     * @param userIdentity The user identity
     * @param maxDays The maximum number of days to search for
     * @return The number of calendar events
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected abstract int internalGetEventsCount(UserIdentity userIdentity, int maxDays) throws MessagingConnectorException;
    
    /**
     * Get emails (no caching)
     * @param userIdentity The user identity
     * @param maxEmails The maximum number of emails to retrieve
     * @return The emails
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected abstract List<EmailMessage> internalGetEmails(UserIdentity userIdentity, int maxEmails) throws MessagingConnectorException;
    
    /**
     * Get the user password for the messaging connector
     * @param userIdentity user to check
     * @return the decrypted user password
     * @throws UserPreferencesException error while reading user preferences
     */
    protected String getUserPassword(UserIdentity userIdentity) throws UserPreferencesException
    {
        if (supportUserCredential())
        {
            String encryptedValue = getUserCryptedPassword(userIdentity);
            return _cryptoHelper.decrypt(encryptedValue);
        }
        else
        {
            throw new MessagingConnectorException("Cannot get password for user " + userIdentity + ": user credential are not supported by this messaging connector", MessagingConnectorException.ExceptionType.CONFIGURATION_EXCEPTION);
        }
    }
    
    /**
     * Get the user password, still crypted
     * @param userIdentity user to check
     * @return the still crypted user password
     * @throws UserPreferencesException error while reading user preferences
     */
    protected String getUserCryptedPassword(UserIdentity userIdentity) throws UserPreferencesException
    {
        return _userPref.getUserPreferenceAsString(userIdentity, "/messaging-connector", Collections.emptyMap(), "messaging-connector-password");
    }
    
    @Override
    public void setUserPassword(UserIdentity userIdentity, String password) throws UserPreferencesException, MessagingConnectorException
    {
        if (supportUserCredential())
        {
            String cryptedPassword = _cryptoHelper.encrypt(password);
            _userPref.addUserPreference(userIdentity, "/messaging-connector", Collections.emptyMap(), "messaging-connector-password", cryptedPassword);
            // Unauthorized cache is invalidated for this user
            _invalidateExceptionForUserInCache(userIdentity, ExceptionType.UNAUTHORIZED);
        }
        else
        {
            throw new MessagingConnectorException("Cannot set password for user " + userIdentity + ": user credential are not supported by this messaging connector", MessagingConnectorException.ExceptionType.CONFIGURATION_EXCEPTION);
        }
    }

    /**
     * Get emails count (no caching)
     * @param userIdentity The user identity
     * @return The emails count
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected abstract int internalGetEmailsCount(UserIdentity userIdentity) throws MessagingConnectorException;
    
    @Override
    public boolean supportInvitation() throws MessagingConnectorException
    {
        return false;
    }

    @Override
    public boolean isEventExist(String eventId, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            return internalIsEventExist(eventId, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }

    /**
     * True if the event exist in the messaging connector
     * @param eventId the event id
     * @param organiser the organiser
     * @return true if the event exist
     * @throws MessagingConnectorException if an error occurred
     */
    protected boolean internalIsEventExist(String eventId, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }
    
    @Override
    public String createEvent(String title, String description, String place, boolean isAllDay, Date startDate, Date endDate, EventRecurrenceTypeEnum recurrenceType, Date untilDate, Map<String, Boolean> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            return internalCreateEvent(title, description, place, isAllDay, startDate, endDate, recurrenceType, untilDate, attendees, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }
    /**
     * Create an event
     * @param title the event title
     * @param description the event description
     * @param place the event place
     * @param isAllDay if the event is all day
     * @param startDate the event start date
     * @param endDate the event end date
     * @param recurrenceType recurrence type
     * @param untilDate until date of the recurring event
     * @param attendees the map of attendees (email -&gt; optional or requested) to set
     * @param organiser the event organiser
     * @return the id of the event created
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected String internalCreateEvent(String title, String description, String place, boolean isAllDay, Date startDate, Date endDate, EventRecurrenceTypeEnum recurrenceType, Date untilDate, Map<String, Boolean> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }

    @Override
    public void updateEvent(String eventId, String title, String description, String place, boolean isAllDay, Date startDate, Date endDate, EventRecurrenceTypeEnum recurrenceType, Date untilDate, Map<String, Boolean> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            internalUpdateEvent(eventId, title, description, place, isAllDay, startDate, endDate, recurrenceType, untilDate, attendees, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }

    /**
     * Update an event
     * @param eventId the event id to delete
     * @param title the event title
     * @param description the event description
     * @param place the event place
     * @param isAllDay if the event is all day
     * @param startDate the event start date
     * @param endDate the event end date
     * @param recurrenceType recurrence type
     * @param untilDate until date of the recurring event
     * @param attendees the map of attendees (email -&gt; optional or requested) to set
     * @param organiser the event organiser
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected void internalUpdateEvent(String eventId, String title, String description, String place, boolean isAllDay, Date startDate, Date endDate, EventRecurrenceTypeEnum recurrenceType, Date untilDate, Map<String, Boolean> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }

    @Override
    public void deleteEvent(String eventId, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            internalDeleteEvent(eventId, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }

    /**
     * Delete an event
     * @param eventId the event id to delete
     * @param organiser the event organiser
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected void internalDeleteEvent(String eventId, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }

    @Override
    public Map<String, AttendeeInformation> getAttendees(String eventId, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            return internalGetAttendees(eventId, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }

    /**
     * Get the map of attendees for an event
     * @param eventId the event id
     * @param organiser the event organiser
     * @return the map of attendees (email -&gt; attendee information)
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected Map<String, AttendeeInformation> internalGetAttendees(String eventId, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }

    @Override
    public void setAttendees(String eventId, Map<String, Boolean> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            internalSetAttendees(eventId, attendees, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }

    /**
     * Set attendees for an event
     * @param eventId the event id
     * @param attendees the map of attendees (email -&gt; optional or requested) to set
     * @param organiser the event organiser
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected void internalSetAttendees(String eventId, Map<String, Boolean> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }

    @Override
    public Map<String, FreeBusyStatus> getFreeBusy(Date startDate, Date endDate, boolean isAllDay, Set<String> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        // Check if one of the last calls returned an exception and throw it directly if needed
        _throwMessagingConnectorExceptionIfInCache(organiser);
        try
        {
            return internalGetFreeBusy(startDate, endDate, isAllDay, attendees, organiser);
        }
        catch (MessagingConnectorException e)
        {
            // Save the exception in cache to avoid to call the server again
            _putExceptionInCache(organiser, e.getType());
            throw e;
        }
    }

    /**
     * Get free/busy status for attendees for a time window
     * @param startDate the start date
     * @param endDate the end date
     * @param isAllDay true if is an allday event
     * @param attendees the list of attendees email
     * @param organiser the event organiser
     * @return the map of attendees (email -&gt; freeBusy status)
     * @throws MessagingConnectorException if failed to get events from server
     */
    protected Map<String, FreeBusyStatus> internalGetFreeBusy(Date startDate, Date endDate, boolean isAllDay, Set<String> attendees, UserIdentity organiser) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }

    @Override
    public boolean userCredentialNeeded()
    {
        boolean credentialNeeded = false;
        if (supportUserCredential())
        {
            UserIdentity user = _currentUserProvider.getUser();
            if (user != null)
            {
                try
                {
                    String password = getUserPassword(user);
                    if (StringUtils.isEmpty(password))
                    {
                        credentialNeeded = true;
                    }
                }
                catch (UserPreferencesException e)
                {
                    credentialNeeded = true;
                }
            }
        }
        return credentialNeeded;
    }

    @Override
    public boolean supportUserCredential()
    {
        return false;
    }
    
    @Override
    public boolean isUserExist(UserIdentity userIdentity) throws MessagingConnectorException
    {
        throw new UnsupportedOperationException("Invitation is not implemented for messaging connector");
    }
    
    private void _invalidateExceptionForUserInCache(UserIdentity userIdentity, MessagingConnectorException.ExceptionType type)
    {
        Set<UserIdentity> usersInCache = _errorCache.getIfPresent(type);
        if (usersInCache != null)
        {
            usersInCache.remove(userIdentity);
        }
    }
    
    private void _putExceptionInCache(UserIdentity userIdentity, MessagingConnectorException.ExceptionType type)
    {
        Cache<ExceptionType, Set<UserIdentity>> cache = null;
        switch (type)
        {
            case TIMEOUT:
                cache = _timeoutErrorCache;
                break;
            case CONFIGURATION_EXCEPTION:
            case UNAUTHORIZED:
            case UNKNOWN:
            default:
                cache = _errorCache;
                break;
        }
        
        Set<UserIdentity> usersInCache = cache.getIfPresent(type);
        if (usersInCache == null)
        {
            usersInCache = new HashSet<>();
            usersInCache.add(userIdentity);
            cache.put(type, usersInCache);
        }
        else
        {
            usersInCache.add(userIdentity);
        }
    }
    
    private MessagingConnectorException.ExceptionType _getExceptionTypeFromCache(UserIdentity userIdentity)
    {
        for (Entry<ExceptionType, Set<UserIdentity>> entry : _errorCache.asMap().entrySet())
        {
            if (entry.getValue().contains(userIdentity))
            {
                // Get the first exception type found for this user (assume that no multiple exception can exist for a same user)
                return entry.getKey();
            }
        }
        
        for (Entry<ExceptionType, Set<UserIdentity>> entry : _timeoutErrorCache.asMap().entrySet())
        {
            if (entry.getValue().contains(userIdentity))
            {
                return entry.getKey();
            }
        }
        
        return null;
    }
    
    private void _throwMessagingConnectorExceptionIfInCache(UserIdentity userIdentity) throws MessagingConnectorException
    {
        MessagingConnectorException.ExceptionType type = _getExceptionTypeFromCache(userIdentity);
        if (type != null)
        {
            throw new MessagingConnectorException(type.name() + " exception was found in cache for user " + userIdentity + ". See previous exception to get the real cause.", type);
        }
    }
    
    /**
     * Internal class for key of events cache
     *
     */
    class EventCacheKey
    {
        private UserIdentity _userIdentity;
        private int _maxDays;
        private int _maxEvents;

        public EventCacheKey (UserIdentity userIdentity, int maxDays, int maxEvents)
        {
            _userIdentity = userIdentity;
            _maxDays = maxDays;
            _maxEvents = maxEvents;
        }
        
        UserIdentity getUserIdentity()
        {
            return _userIdentity;
        }
        
        int getMaxDays()
        {
            return _maxDays;
        }
        
        int getMaxEvents()
        {
            return _maxEvents;
        }
        
        @Override
        public int hashCode()
        {
            return Objects.hash(_userIdentity, _maxDays, _maxEvents);
        }
        
        @Override
        public boolean equals(Object obj)
        {
            if (obj == null)
            {
                return false;
            }
            
            if (!(obj instanceof EventCacheKey))
            {
                return false;
            }
            
            EventCacheKey toCompare = (EventCacheKey) obj;
            
            return _userIdentity.equals(toCompare.getUserIdentity()) && _maxDays == toCompare.getMaxDays() && _maxEvents == toCompare.getMaxEvents();
        }
    }
    
    /**
     * Internal class for key of events count cache
     *
     */
    class EventCountCacheKey
    {
        private UserIdentity _userIdentity;
        private int _maxDays;

        public EventCountCacheKey (UserIdentity userIdentity, int maxDays)
        {
            _userIdentity = userIdentity;
            _maxDays = maxDays;
        }
        
        UserIdentity getUserIdentity()
        {
            return _userIdentity;
        }
        
        int getMaxDays()
        {
            return _maxDays;
        }
        
        
        @Override
        public int hashCode()
        {
            return Objects.hash(_userIdentity, _maxDays);
        }
        
        @Override
        public boolean equals(Object obj)
        {
            if (obj == null)
            {
                return false;
            }
            
            if (!(obj instanceof EventCountCacheKey))
            {
                return false;
            }
            
            EventCountCacheKey toCompare = (EventCountCacheKey) obj;
            
            return _userIdentity.equals(toCompare.getUserIdentity()) && _maxDays == toCompare.getMaxDays();
        }
    }
    
    /**
     * Internal class for key of events count cache
     *
     */
    class EmailCacheKey
    {
        private UserIdentity _userIdentity;
        private int _maxEmails;

        public EmailCacheKey (UserIdentity userIdentity, int maxEmails)
        {
            _userIdentity = userIdentity;
            _maxEmails = maxEmails;
        }
        
        UserIdentity getUserIdentity()
        {
            return _userIdentity;
        }

        int getMaxEmails()
        {
            return _maxEmails;
        }
        
        @Override
        public int hashCode()
        {
            return Objects.hash(_userIdentity, _maxEmails);
        }
        
        @Override
        public boolean equals(Object obj)
        {
            if (obj == null)
            {
                return false;
            }
            
            if (!(obj instanceof EmailCacheKey))
            {
                return false;
            }
            
            EmailCacheKey toCompare = (EmailCacheKey) obj;
            
            return _userIdentity.equals(toCompare.getUserIdentity()) && _maxEmails == toCompare.getMaxEmails();
        }
    }

}