/*
 * The contents of this file are subject to the terms
 * of the Common Development and Distribution License
 * (the License).  You may not use this file except in
 * compliance with the License.
 *
 * You can obtain a copy of the license at
 * https://glassfish.dev.java.net/public/CDDLv1.0.html or
 * glassfish/bootstrap/legal/CDDLv1.0.txt.
 * See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL
 * Header Notice in each file and include the License file
 * at glassfish/bootstrap/legal/CDDLv1.0.txt.
 * If applicable, add the following below the CDDL Header,
 * with the fields enclosed by brackets [] replaced by
 * you own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 *
 * Copyright (c) Ericsson AB, 2004-2007. All rights reserved.
 * Copyright 1997-2007 Sun Microsystems, Inc. All rights reserved.
 */
package org.jvnet.glassfish.comms.replication.sessmgmt;

import java.io.IOException;
import java.io.Serializable;
import java.security.AccessController;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;
import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.servlet.sip.Address;
import javax.servlet.sip.TimerListener;
import com.ericsson.ssa.sip.DialogSet;
import com.ericsson.ssa.sip.PathNode.Type;
import com.ericsson.ssa.sip.PersistentSipSessionManagerBase;
import com.ericsson.ssa.sip.SipApplicationSessionImpl;
import com.ericsson.ssa.sip.SipApplicationSessionUtil;
import com.ericsson.ssa.sip.SipSessionDialogImpl;
import com.ericsson.ssa.sip.timer.ServletTimerImpl;
import com.sun.appserv.ha.spi.BackingStore;
import com.sun.appserv.ha.spi.BackingStoreException;
import com.sun.appserv.ha.spi.BackingStoreFactory;
import com.sun.appserv.ha.spi.BackingStoreRegistry;
import com.sun.appserv.ha.spi.SimpleMetadata;
import com.sun.ejb.containers.util.cache.BaseCache;
import com.sun.enterprise.ee.web.sessmgmt.EEPersistenceTypeResolver;
import com.sun.enterprise.ee.web.sessmgmt.JxtaBackingStoreFactory;
import com.sun.enterprise.ee.web.sessmgmt.JxtaBackingStoreImpl;
import com.sun.enterprise.ee.web.sessmgmt.JxtaReplicationReceiver;
import com.sun.enterprise.ee.web.sessmgmt.JxtaReplicationSender;
import com.sun.enterprise.ee.web.sessmgmt.PurgeCapable;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationHealthChecker;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationManager;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationMessageRouter;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationResponseRepository;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationState;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationStateQueryResponse;
import com.sun.enterprise.ee.web.sessmgmt.ReplicationUtil;
import com.sun.enterprise.ee.web.sessmgmt.StorePool;
import com.sun.enterprise.ee.web.sessmgmt.StorePoolElement;
import com.sun.logging.LogDomains;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.security.SecurityUtil;
import org.jvnet.glassfish.comms.util.LogUtil;

public class SipTransactionPersistentManager 
        extends PersistentSipSessionManagerBase
        implements ReplicationManager, PurgeCapable {

    private static final Logger _logger = Logger.getLogger(LogUtil.SIP_LOG_DOMAIN);
    protected static final String MODE_SIP = ReplicationState.MODE_SIP;
    final static String DUPLICATE_IDS_SEMANTICS_PROPERTY 
        = ReplicationState.DUPLICATE_IDS_SEMANTICS_PROPERTY; 
    final static String COMMAND_MAP = "commandmap";
    final static String SUPPRESS_LOAD_ACK_PROPERTY 
        = ReplicationState.SUPPRESS_LOAD_ACK_PROPERTY;
    final static String SIP_TYPE_SAS = "sip_type_sas";
    private static final Level TRACE_LEVEL = Level.FINE;
    protected static int _maxBaseCacheSize = 4096;
    protected static float _loadFactor = 0.75f;
    
    private static final Map sipApplicationSessionCommandMap = new HashMap();
    private static final Map sipSessionCommandMap = new HashMap();
    private static final Map servletTimerCommandMap = new HashMap();
    
    public final static String SAVE_COMMAND 
        = ReplicationState.SAVE_COMMAND;
    public final static String VALVE_SAVE_COMMAND 
        = ReplicationState.VALVE_SAVE_COMMAND;    
    public final static String REMOVE_COMMAND 
        = ReplicationState.REMOVE_COMMAND;    
    public final static String UNDEPLOY_COMMAND 
        = ReplicationState.UNDEPLOY_COMMAND;
    public final static String REMOVE_EXPIRED_COMMAND 
        = ReplicationState.REMOVE_EXPIRED_COMMAND;
    public final static String UPDATE_LAST_ACCESS_TIME_COMMAND 
        = ReplicationState.UPDATE_LAST_ACCESS_TIME_COMMAND;
    public final static String MESSAGE_BROADCAST_QUERY 
        = ReplicationState.MESSAGE_BROADCAST_QUERY;
    public final static String MESSAGE_BROADCAST_LOAD_RECEIVED
        = ReplicationState.MESSAGE_BROADCAST_LOAD_RECEIVED;
    public final static String MESSAGE_BROADCAST_LOAD_ADVISORY
        = ReplicationState.MESSAGE_BROADCAST_LOAD_ADVISORY;
    public final static String MESSAGE_BROADCAST_PURGE_ADVISORY
        = ReplicationState.MESSAGE_BROADCAST_PURGE_ADVISORY;    
    
    public final static String SAVE_SAS_COMMAND = "savesas";
    public final static String REMOVE_SAS_COMMAND = "removesas";    
    public final static String UPDATE_LAST_ACCESS_TIME_SAS_COMMAND 
        = "updatelastaccesstimesas";
    public final static String MESSAGE_BROADCAST_QUERY_SAS 
        = "broadcastfindsas";
    public final static String MESSAGE_BROADCAST_LOAD_RECEIVED_SAS
        = "broadcastloadreceivedsas";
    public final static String MESSAGE_BROADCAST_LOAD_ADVISORY_SAS
        = "broadcastloadadvisorysas";    
    
    public final static String SAVE_SIP_SESSION_COMMAND = "savesipsession";
    public final static String REMOVE_SIP_SESSION_COMMAND = "removesipsession";    
    public final static String UPDATE_LAST_ACCESS_TIME_SIP_SESSION_COMMAND 
        = "updatelastaccesstimesipsession";
    public final static String MESSAGE_BROADCAST_QUERY_SIP_SESSION 
        = "broadcastfindsipsession"; 
    public final static String MESSAGE_BROADCAST_LOAD_RECEIVED_SIP_SESSION
        = "broadcastloadreceivedsipsession";
    public final static String MESSAGE_BROADCAST_LOAD_ADVISORY_SIP_SESSION
        = "broadcastloadadvisorysipsession";    
    
    public final static String SAVE_SERVLET_TIMER_COMMAND 
        = "saveservlettimer";
    public final static String REMOVE_SERVLET_TIMER_COMMAND 
        = "removeservlettimer";    
    public final static String UPDATE_LAST_ACCESS_TIME_SERVLET_TIMER_COMMAND 
        = "updatelastaccesstimeservlettimer";
    public final static String MESSAGE_BROADCAST_QUERY_SERVLET_TIMER 
        = "broadcastfindservlettimer";
    public final static String MESSAGE_BROADCAST_LOAD_RECEIVED_SERVLET_TIMER
        = "broadcastloadreceivedservlettimer";
    public final static String MESSAGE_BROADCAST_LOAD_ADVISORY_SERVLET_TIMER
        = "broadcastloadadvisoryservlettimer";    
    
    /**
     * the list of method names that are broadcasts
     */ 
    private static List broadcastMethods 
        = Arrays.asList(             
            MESSAGE_BROADCAST_QUERY_SAS,
            MESSAGE_BROADCAST_LOAD_RECEIVED_SAS,
            MESSAGE_BROADCAST_LOAD_ADVISORY_SAS,
            MESSAGE_BROADCAST_QUERY_SIP_SESSION,
            MESSAGE_BROADCAST_LOAD_RECEIVED_SIP_SESSION,
            MESSAGE_BROADCAST_LOAD_ADVISORY_SIP_SESSION,
            MESSAGE_BROADCAST_QUERY_SERVLET_TIMER,
            MESSAGE_BROADCAST_LOAD_RECEIVED_SERVLET_TIMER,
            MESSAGE_BROADCAST_LOAD_ADVISORY_SERVLET_TIMER);    
    
    static {
        checkSessionCacheProperties();
        initializeCommandMaps();
        registerBroadcastMethods();
    }

    private SessionFactory sessionFactory;
    private StorePool sipApplicationSessionStorePool;
    private StorePool sipSessionStorePool;
    private StorePool servletTimerStorePool;
    private final String instanceName;
    private ReplicationMessageRouter router;

    /**
     * Our cache of replicated SipApplicationSession ReplicationState objects
     * keyed by id
     */
    protected BaseCache replicatedSipApplicationSessions = new BaseCache();

    /**
     * Our cache of replicated SipSession ReplicationState objects
     * keyed by id
     */
    protected BaseCache replicatedSipSessions = new BaseCache();
    
    /**
     * Our cache of replicated ServletTimer ReplicationState objects
     * keyed by id
     */
    protected BaseCache replicatedServletTimers = new BaseCache(); 

    /**
     * Our cache of replicated SipApplicationSession ReplicationStateUpdate
     * objects
     * keyed by id
     */
    protected BaseCache replicatedSipApplicationSessionUpdates = null;

    /**
     * Our cache of replicated SipSession ReplicationStateUpdate objects
     * keyed by id
     */
    protected BaseCache replicatedSipSessionUpdates = null;
    
    /**
     * Our cache of replicated ServletTimer ReplicationStateUpdate objects
     * keyed by id
     */
    protected BaseCache replicatedServletTimerUpdates = null;
    
    /**
    * Our Replicator instance for SipApplicationSessions (for SimpleMetadata)
    */
    protected BackingStore sipApplicationSessionBackingStore = null; 
    
    /**
    * Our Replicator instance SipSessions (for SimpleMetadata)
    */
    protected BackingStore sipSessionBackingStore = null; 
    
    /**
    * Our Replicator instance for ServletTimers (for SimpleMetadata)
    */
    protected BackingStore servletTimerBackingStore = null;     
    
    protected String _passedInPersistenceType = null;
    protected boolean _duplicateIdsSemanticsAllowed = false;     

    private ActivationHelper activationHelper;

    /** Creates a new instance of ReplicationManagerBase */
    public SipTransactionPersistentManager() {
        super();
        //initialize replicated sessions cache
        replicatedSipApplicationSessions = new BaseCache();
        replicatedSipApplicationSessions.init(_maxBaseCacheSize, _loadFactor,
                                              null);
        replicatedSipSessions = new BaseCache();
        replicatedSipSessions.init(_maxBaseCacheSize, _loadFactor, null);
        replicatedServletTimers = new BaseCache();
        replicatedServletTimers.init(_maxBaseCacheSize, _loadFactor, null);
        
        replicatedSipApplicationSessionUpdates = new BaseCache();
        replicatedSipApplicationSessionUpdates.init(_maxBaseCacheSize,
                                                    _loadFactor, null); 
        replicatedSipSessionUpdates = new BaseCache();
        replicatedSipSessionUpdates.init(_maxBaseCacheSize, _loadFactor, null); 
        replicatedServletTimerUpdates = new BaseCache();
        replicatedServletTimerUpdates.init(_maxBaseCacheSize, _loadFactor, null); 
        instanceName = ReplicationUtil.getInstanceName();

        activationHelper = new ActivationHelper(this);

        router = ReplicationMessageRouter.createInstance();
    }  

    protected static boolean checkSessionCacheProperties() {
        boolean result = false;

        try {
            Properties props = System.getProperties();
            String cacheSize = props.getProperty(
                    "HTTP_SESSION_CACHE_MAX_BASE_CACHE_SIZE");

            if (null != cacheSize) {
                _maxBaseCacheSize = (new Integer(cacheSize).intValue());
            }

            String loadFactor = props.getProperty(
                    "HTTP_SESSION_CACHE_MAX_BASE_LOAD_FACTOR");

            if (null != loadFactor) {
                _loadFactor = (new Float(loadFactor).floatValue());
            }
        } catch (Exception e) {
            //do nothing accept defaults
        }

        return result;
    }
    
    protected static void initializeCommandMaps() {
        sipApplicationSessionCommandMap.put(SAVE_COMMAND, SAVE_SAS_COMMAND);
        sipSessionCommandMap.put(SAVE_COMMAND, SAVE_SIP_SESSION_COMMAND);
        servletTimerCommandMap.put(SAVE_COMMAND, SAVE_SERVLET_TIMER_COMMAND); 

        sipApplicationSessionCommandMap.put(VALVE_SAVE_COMMAND, SAVE_SAS_COMMAND);
        sipSessionCommandMap.put(VALVE_SAVE_COMMAND, SAVE_SIP_SESSION_COMMAND);
        servletTimerCommandMap.put(VALVE_SAVE_COMMAND, SAVE_SERVLET_TIMER_COMMAND);        
        
        sipApplicationSessionCommandMap.put(REMOVE_COMMAND, REMOVE_SAS_COMMAND);
        sipSessionCommandMap.put(REMOVE_COMMAND, REMOVE_SIP_SESSION_COMMAND);
        servletTimerCommandMap.put(REMOVE_COMMAND, REMOVE_SERVLET_TIMER_COMMAND);  
        
        sipApplicationSessionCommandMap.put(UPDATE_LAST_ACCESS_TIME_COMMAND, UPDATE_LAST_ACCESS_TIME_SAS_COMMAND);
        sipSessionCommandMap.put(UPDATE_LAST_ACCESS_TIME_COMMAND, UPDATE_LAST_ACCESS_TIME_SIP_SESSION_COMMAND);
        servletTimerCommandMap.put(UPDATE_LAST_ACCESS_TIME_COMMAND, UPDATE_LAST_ACCESS_TIME_SERVLET_TIMER_COMMAND);
        
        sipApplicationSessionCommandMap.put(MESSAGE_BROADCAST_QUERY, MESSAGE_BROADCAST_QUERY_SAS);
        sipSessionCommandMap.put(MESSAGE_BROADCAST_QUERY, MESSAGE_BROADCAST_QUERY_SIP_SESSION);
        servletTimerCommandMap.put(MESSAGE_BROADCAST_QUERY, MESSAGE_BROADCAST_QUERY_SERVLET_TIMER);        
    } 
    
    protected static void registerBroadcastMethods() {
        ReplicationMessageRouter router 
            = ReplicationMessageRouter.createInstance();
        router.registerBroadcastMethodList(broadcastMethods);
    }
    
    public void start() throws LifecycleException {
        super.start();
        router.addReplicationManager(this.getApplicationId(), this);
    }
    
    /**
     * Releases any resources held by this SipSessionManager.
     */
    public void release() {
        router.removeReplicationManager(this.getApplicationId());
    }
    
    /**
    * get the backingStore for SipApplicationSession
    */ 
    public BackingStore getSipApplicationSessionBackingStore() {
        if(sipApplicationSessionBackingStore == null) {
            sipApplicationSessionBackingStore = this.createBackingStore(sipApplicationSessionCommandMap, SIP_TYPE_SAS);
        }
        return sipApplicationSessionBackingStore;
    }
    
    /**
    * get the backingStore for SipSession
    */ 
    public BackingStore getSipSessionBackingStore() {
        if(sipSessionBackingStore == null) {
            sipSessionBackingStore = this.createBackingStore(sipSessionCommandMap, null);           
        }
        return sipSessionBackingStore;
    } 
    
    /**
    * get the backingStore for ServletTimer
    */ 
    public BackingStore getServletTimerBackingStore() {
        if(servletTimerBackingStore == null) {
            servletTimerBackingStore = this.createBackingStore(servletTimerCommandMap, null);            
        }
        return servletTimerBackingStore;
    }
    
    /**
    * create and set the backing store
    * @param commandMap map used to translate commands
    */     
    BackingStore createBackingStore(Map commandMap, String sipType) {
        BackingStoreFactory storeFactory = new JxtaBackingStoreFactory();
        BackingStoreRegistry backingStoreRegistry 
            = BackingStoreRegistry.getInstance();
        Properties inputEnv 
            = backingStoreRegistry.getFactoryClassEnv(getPassedInPersistenceType());
        Properties env = (Properties)inputEnv.clone();        
        //does this manager & backing store support duplicate id semantics
        //for batch replication usage
        env.put(DUPLICATE_IDS_SEMANTICS_PROPERTY, Boolean.valueOf(this.isDuplicateIdsSemanticsAllowed()));
        env.put(COMMAND_MAP, commandMap);
        if(sipType != null && sipType.equals(SIP_TYPE_SAS)) {
            env.put(SUPPRESS_LOAD_ACK_PROPERTY, Boolean.TRUE);
        }
        BackingStore backingStore = null;
        try {
            backingStore = storeFactory.createBackingStore(
                        SimpleMetadata.class,     //type
                        this.getApplicationId(), //appid
                        env);
        } catch (BackingStoreException ex) {
            //deliberate no-op
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("backingStore: " + backingStore);
        }         
        if(backingStore != null) {
            if(backingStore instanceof JxtaBackingStoreImpl) {
                ((JxtaBackingStoreImpl)backingStore).setMode(MODE_SIP);
            }        
        }
        return backingStore;
    }        
    
    public boolean isDuplicateIdsSemanticsAllowed() {
        return _duplicateIdsSemanticsAllowed;
    }    
    
    public void setDuplicateIdsSemanticsAllowed(boolean value) {
        _duplicateIdsSemanticsAllowed = value;
    }   
    
    public String getPassedInPersistenceType() {
        return _passedInPersistenceType;
    }    
    
    public void setPassedInPersistenceType(String persistenceType) {
        _passedInPersistenceType = persistenceType;
    }    
    
    //SAS cache methods

    /**
     * get the replicated SipApplicationSessions cache
     */ 
    public BaseCache getReplicatedSipApplicationSessions() {
        return replicatedSipApplicationSessions;
    }
    
    /**
     * set the replicated SipApplicationSessions cache
     * @param sesstable
     */ 
    public void setReplicatedSipApplicationSessions(BaseCache sesstable) {
        replicatedSipApplicationSessions = sesstable;
    } 
    
    /**
     * Put session State in SipApplicationSession replica cache
     * @param sessionState
     */    
    protected synchronized void putInSipApplicationSessionReplicationCache(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName()
                         + ">>putInSipApplicationSessionReplicationCache id: "
                         + sessionState.getId());
        } 
        String id = (String)sessionState.getId();
        if(id == null) {
            return;
        }
        
        ReplicationState currentState = (ReplicationState)replicatedSipApplicationSessions.get(id);
        if(_logger.isLoggable(Level.FINE)) {
            if(currentState != null) {
                _logger.fine("currentVersion: " + currentState.getVersion()
                             + " newVersion: " + sessionState.getVersion());
            }
        }
        if((currentState != null)
                && currentState.getVersion() > sessionState.getVersion()) {
            return;
        }
        replicatedSipApplicationSessions.put(sessionState.getId(), sessionState);
        if(_logger.isLoggable(Level.FINE)) {
            _logger.log(Level.FINE, "in " + this.getClass().getName()
                + ">>putInSipApplicationSessionReplicationCache complete id: "
                + sessionState.getId() + "[ver:" + sessionState.getVersion()
                + "]");
        }        
    }
    
    /**
     * get SipApplicationSession State from replica cache based on the id
     * @param id
     */    
    protected ReplicationState getFromSipApplicationSessionReplicationCache(String id) {
        ReplicationState returnState 
            = (ReplicationState)replicatedSipApplicationSessions.get(id);
        if(returnState != null) {
            ReplicationStateUpdate returnStateUpdate
                = this.getFromSipApplicationSessionReplicationUpdateCache(id);
            //apply update if it exists and has higher version
            if(returnStateUpdate != null) {
                if(returnStateUpdate.getVersion() > returnState.getVersion()) {
                    returnState.setVersion(returnStateUpdate.getVersion());
                    returnState.setLastAccess(returnStateUpdate.getLastAccess());
                    returnState.setContainerExtraParamsState(returnStateUpdate.getContainerExtraParams());                    
                }
                this.removeFromSipApplicationSessionReplicationUpdateCache(returnStateUpdate.getId());
            }            
        }
        return returnState;
    }
    
    /**
     * remove session State from replica cache based on the id 
     * @param id
     */    
    protected ReplicationState removeFromSipApplicationSessionReplicationCache(String id) {
        if(id == null) {
            return null;
        }
        removeFromSipApplicationSessionReplicationUpdateCache(id);
        return (ReplicationState)replicatedSipApplicationSessions.remove(id);        
    }
    
    //SS cache methods
    
    /**
    * get the replicated SipSessions cache
    */ 
    public BaseCache getReplicatedSipSessions() {
        return replicatedSipSessions;
    }
    
    /**
    * set the replicated SipSessions cache
    * @param sesstable
    */ 
    public void setReplicatedSipSessions(BaseCache sesstable) {
        replicatedSipSessions = sesstable;
    }
    
    /**
     * Put session State in SipApplicationSession replica cache
     * @param sessionState
     */    
    protected synchronized void putInSipSessionReplicationCache(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName()
                         + ">>putInSipSessionReplicationCache id: "
                         + sessionState.getId());
        } 
        String id = (String)sessionState.getId();
        if(id == null) {
            return;
        }
        
        ReplicationState currentState = (ReplicationState)replicatedSipSessions.get(id);
        if(_logger.isLoggable(Level.FINE)) {
            if(currentState != null) {
                _logger.fine("currentVersion: " + currentState.getVersion()
                             + " newVersion: " + sessionState.getVersion());
            }
        }
        if((currentState != null) && currentState.getVersion() > sessionState.getVersion()) {
            return;
        }
        replicatedSipSessions.put(sessionState.getId(), sessionState);
        if(_logger.isLoggable(Level.FINE)) {
            _logger.log(Level.FINE, "in " + this.getClass().getName()
                        + ">>putInSipSessionReplicationCache complete id: "
                        + sessionState.getId() + "[ver:"
                        + sessionState.getVersion() + "]");
        }        
    } 
    
    /**
     * get SipSession State from replica cache based on the id
     * @param id
     */    
    protected ReplicationState getFromSipSessionReplicationCache(String id) {
        ReplicationState returnState 
            = (ReplicationState)replicatedSipSessions.get(id);
        if(returnState != null) {
            ReplicationStateUpdate returnStateUpdate
                = this.getFromSipSessionReplicationUpdateCache(id);
            //apply update if it exists and has higher version
            if(returnStateUpdate != null) {
                if(returnStateUpdate.getVersion() > returnState.getVersion()) {
                    returnState.setVersion(returnStateUpdate.getVersion());
                    returnState.setLastAccess(returnStateUpdate.getLastAccess());
                    returnState.setContainerExtraParamsState(returnStateUpdate.getContainerExtraParams());                    
                }
                this.removeFromSipSessionReplicationUpdateCache(returnStateUpdate.getId());
            }            
        }
        return returnState;
    }
    
    /**
     * remove session State from replica cache based on the id 
     * @param id
     */    
    protected ReplicationState removeFromSipSessionReplicationCache(String id) {
        if(id == null) {
            return null;
        }
        removeFromSipSessionReplicationUpdateCache(id);
        return (ReplicationState)replicatedSipSessions.remove(id);        
    }    
    
 //ServletTimer cache methods
    
    /**
    * get the replicated ServletTimers cache
    */ 
    public BaseCache getReplicatedServletTimers() {
        return replicatedServletTimers;
    }
    
    /**
    * set the replicated ServletTimers cache
    * @param sesstable
    */ 
    public void setReplicatedServletTimers(BaseCache sesstable) {
        replicatedServletTimers = sesstable;
    }
    
    /**
     * Put session timer State in ServletTimers replica cache
     * @param sessionState
     */    
    protected synchronized void putInServletTimerReplicationCache(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName()
                         + ">>putInServletTimerReplicationCache id: "
                         + sessionState.getId());
        } 
        String id = (String)sessionState.getId();
        if(id == null) {
            return;
        }
        
        ReplicationState currentState = (ReplicationState)replicatedServletTimers.get(id);
        if(_logger.isLoggable(Level.FINE)) {
            if(currentState != null) {
                _logger.fine("currentVersion: " + currentState.getVersion()
                             + " newVersion: " + sessionState.getVersion());
            }
        }
        if((currentState != null) && currentState.getVersion() > sessionState.getVersion()) {
            return;
        }
        replicatedServletTimers.put(sessionState.getId(), sessionState);
        if(_logger.isLoggable(Level.FINE)) {
            _logger.log(Level.FINE, "in " + this.getClass().getName()
                        + ">>putInSipSessionReplicationCache complete id: "
                        + sessionState.getId() + "[ver:"
                        + sessionState.getVersion() + "]");
        }        
    } 
    
    /**
     * get ServletTimer State from replica cache based on the id
     * @param id
     */    
    protected ReplicationState getFromServletTimerReplicationCache(String id) {
        ReplicationState returnState 
            = (ReplicationState)replicatedSipSessions.get(id);
        if(returnState != null) {
            ReplicationStateUpdate returnStateUpdate
                = this.getFromServletTimerReplicationUpdateCache(id);
            //apply update if it exists and has higher version
            if(returnStateUpdate != null) {
                if(returnStateUpdate.getVersion() > returnState.getVersion()) {
                    returnState.setVersion(returnStateUpdate.getVersion());
                    returnState.setLastAccess(returnStateUpdate.getLastAccess());
                    returnState.setContainerExtraParamsState(returnStateUpdate.getContainerExtraParams());
                }
                this.removeFromServletTimerReplicationUpdateCache(returnStateUpdate.getId());
            }            
        }
        return returnState;
    }
    
    /**
     * remove session State from replica cache based on the id 
     * @param id
     */    
    protected ReplicationState removeFromServletTimerReplicationCache(String id) {
        if(id == null) {
            return null;
        }
        removeFromServletTimerReplicationUpdateCache(id);
        return (ReplicationState)replicatedServletTimers.remove(id);        
    }    
    
    //SAS updates cache methods    
    
    /**
     * Put SipApplicationSession State updates in replica cache
     * @param sessionState
     */    
    protected synchronized void putInSipApplicationSessionReplicationUpdateCache(ReplicationStateUpdate sessionStateUpdate) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName()
                         + ">>putInSipApplicationSessionUpdateCache id: "
                         + sessionStateUpdate.getId());
        }
        if(sessionStateUpdate == null) {
            return;
        }
        String id = (String)sessionStateUpdate.getId();
        if(id == null) {
            return;
        }
        
        ReplicationStateUpdate currentStateUpdate = this.getFromSipApplicationSessionReplicationUpdateCache(id);
        if(_logger.isLoggable(Level.FINE)) {
            if(currentStateUpdate != null) {
                _logger.fine("currentVersion: "
                             + currentStateUpdate.getVersion()
                             + " newVersion: "
                             + sessionStateUpdate.getVersion());
            }
        }
        if((currentStateUpdate != null) && currentStateUpdate.getVersion() > sessionStateUpdate.getVersion()) {
            return;
        }
        replicatedSipApplicationSessionUpdates.put(sessionStateUpdate.getId(), sessionStateUpdate);
    }
    
    /**
     * get SipApplicationSession State update from replica cache based on the
     * id
     *
     * @param id
     */    
    protected ReplicationStateUpdate getFromSipApplicationSessionReplicationUpdateCache(String id) {    
        return (ReplicationStateUpdate)replicatedSipApplicationSessionUpdates.get(id);
    }
    
    /**
     * remove SipApplicationSession State update from replica cache based on
     * the id of sessionStateUpdate
     * @param id
     */    
    protected void removeFromSipApplicationSessionReplicationUpdateCache(String id) {
        if(id == null) {
            return;
        }
        replicatedSipApplicationSessionUpdates.remove(id);
    }
    
    //SS updates cache methods
    
    /**
     * Put SipSession State updates in replica cache
     * @param sessionState
     */    
    protected synchronized void putInSipSessionReplicationUpdateCache(ReplicationStateUpdate sessionStateUpdate) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName()
                         + ">>putInSipSessionReplicationUpdateCache id: "
                         + sessionStateUpdate.getId());
        }
        if(sessionStateUpdate == null) {
            return;
        }
        String id = (String)sessionStateUpdate.getId();
        if(id == null) {
            return;
        }
        
        ReplicationStateUpdate currentStateUpdate = this.getFromSipSessionReplicationUpdateCache(id);
        if(_logger.isLoggable(Level.FINE)) {
            if(currentStateUpdate != null) {
                _logger.fine("currentVersion: "
                             + currentStateUpdate.getVersion()
                             + " newVersion: "
                             + sessionStateUpdate.getVersion());
            }
        }
        if((currentStateUpdate != null)
                && currentStateUpdate.getVersion() > sessionStateUpdate.getVersion()) {
            return;
        }
        replicatedSipSessionUpdates.put(sessionStateUpdate.getId(),
                                        sessionStateUpdate);
    }
    
    /**
     * get SipSession State update from replica cache based on the id
     * @param id
     */    
    protected ReplicationStateUpdate getFromSipSessionReplicationUpdateCache(String id) {    
        return (ReplicationStateUpdate)replicatedSipSessionUpdates.get(id);
    }
    
    /**
     * remove SipSession State update from replica cache based on the id of
     * sessionStateUpdate
     * @param id
     */    
    protected void removeFromSipSessionReplicationUpdateCache(String id) {
        if(id == null) {
            return;
        }
        replicatedSipSessionUpdates.remove(id);
    }   
    
    //ServletTimer updates cache methods
    
    /**
     * Put ServletTimer State updates in replica cache
     * @param sessionState
     */    
    protected synchronized void putInServletTimerReplicationUpdateCache(ReplicationStateUpdate sessionStateUpdate) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName()
                         + ">>putInServletTimerReplicationUpdateCache id: "
                         + sessionStateUpdate.getId());
        }
        if(sessionStateUpdate == null) {
            return;
        }
        String id = (String)sessionStateUpdate.getId();
        if(id == null) {
            return;
        }
        
        ReplicationStateUpdate currentStateUpdate = this.getFromServletTimerReplicationUpdateCache(id);
        if(_logger.isLoggable(Level.FINE)) {
            if(currentStateUpdate != null) {
                _logger.fine("currentVersion: "
                             + currentStateUpdate.getVersion()
                             + " newVersion: "
                             + sessionStateUpdate.getVersion());
            }
        }
        if((currentStateUpdate != null)
                && currentStateUpdate.getVersion() > sessionStateUpdate.getVersion()) {
            return;
        }
        replicatedServletTimerUpdates.put(sessionStateUpdate.getId(),
                                        sessionStateUpdate);
    }
    
    /**
     * get ServletTimer State update from replica cache based on the id
     * @param id
     */    
    protected ReplicationStateUpdate getFromServletTimerReplicationUpdateCache(String id) {    
        return (ReplicationStateUpdate)replicatedServletTimerUpdates.get(id);
    }
    
    /**
     * remove ServletTimer State update from replica cache based on the id of
     * sessionStateUpdate
     * @param id
     */    
    protected void removeFromServletTimerReplicationUpdateCache(String id) {
        if(id == null) {
            return;
        }
        replicatedServletTimerUpdates.remove(id);
    }
    
    private SipApplicationSessionExtraParams deserializeSipApplicationSessionExtraParams(String id) 
        throws IOException, ClassNotFoundException {
        ReplicationState sasState 
            = this.getFromSipApplicationSessionReplicationCache(id);
        ReplicationStateUpdate sasUpdateState 
            = this.getFromSipApplicationSessionReplicationUpdateCache(id);
        if(sasState == null) {
            return null;
        }
        if(sasUpdateState == null || sasState.getVersion() > sasUpdateState.getVersion()) {
            if(sasState.getContainerExtraParamsState() == null) {
                return null;
            } else {
                return (SipApplicationSessionExtraParams)ReplicationState.getObjectValue(sasState.getContainerExtraParamsState());
            }
        } else {
            if(sasUpdateState.getContainerExtraParams() == null) {
                return null;
            } else {
                return (SipApplicationSessionExtraParams)ReplicationState.getObjectValue(sasUpdateState.getContainerExtraParams());
            }
        }
    }    
    
    private ServletTimerExtraParams deserializeServletTimerExtraParams(String id) 
        throws IOException, ClassNotFoundException {
        ReplicationState timerState 
            = this.getFromServletTimerReplicationCache(id);
        ReplicationStateUpdate timerUpdateState 
            = this.getFromServletTimerReplicationUpdateCache(id);
        if(timerState == null) {
            return null;
        }
        if(timerUpdateState == null || timerState.getVersion() > timerUpdateState.getVersion()) {
            if(timerState.getContainerExtraParamsState() == null) {
                return null;
            } else {
                return (ServletTimerExtraParams)ReplicationState.getObjectValue(timerState.getContainerExtraParamsState());
            }
        } else {
            if(timerUpdateState.getContainerExtraParams() == null) {
                return null;
            } else {
                return (ServletTimerExtraParams)ReplicationState.getObjectValue(timerUpdateState.getContainerExtraParams());
            }
        }
    }
    
    //end cache methods
    
    //start receiver side process methods
    
    //SipApplicationSession process methods

    /**
     * process the save of a SipApplicationSession
     * @param sessionState - contains the save command
     */     
    public void processSavesas(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processSavesas");
        }        
        this.putInSipApplicationSessionReplicationCache(sessionState);
    }    

    /**
     * process the removal of a SipApplicationSession
     * @param sessionState - contains the remove command
     */     
    public void processRemovesas(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processRemovesas");
        } 
        if(sessionState == null) {
            return;
        }
        replicatedSipApplicationSessions.remove(sessionState.getId());
        removeFromSipApplicationSessionReplicationUpdateCache((String)sessionState.getId());      
    }

    /**
     * process the update of a SipApplicationSession
     * @param sessionState - contains the updatelastaccesstime command
     */    
    public void processUpdatelastaccesstimesas(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processUpdatelastaccesstimesas");
            _logger.fine("processUpdatelastaccesstimesas:version:" + sessionState.getVersion());
        }
        String id = (String)sessionState.getId();
        ReplicationState state = this.getFromSipApplicationSessionReplicationCache(id);
        if(state != null) {
            state.setLastAccess(sessionState.getLastAccess());
            state.setVersion(sessionState.getVersion());
            state.setContainerExtraParamsState(sessionState.getContainerExtraParamsState());
            this.putInSipApplicationSessionReplicationCache(state);
        } else {
            if(_logger.isLoggable(Level.FINE)) {                
                _logger.fine("processUpdatelastaccesstimesas: attempting to update a session not yet stored:id:" + sessionState.getId());
            }
            this.putInSipApplicationSessionReplicationUpdateCache(new ReplicationStateUpdate(id, sessionState.getVersion(), sessionState.getLastAccess(), sessionState.getContainerExtraParamsState()) );
        }        
    }
    
    /**
     * process the broadcastfindsas for SipApplicationSession
     * @param queryState
     */     
    public ReplicationState processBroadcastfindsas(ReplicationState queryState) {
        //complete query and send back response
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindsas:instance: " + instanceName);
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindsas:id=" + queryState.getId());                        
        } 
        
        //look in active cache
        HASipApplicationSession haSipApplicationSession = null;
        SipApplicationSessionImpl sipApplicationSession 
            = findSipApplicationSessionFromCacheOnly((String)queryState.getId());
        if(sipApplicationSession instanceof HASipApplicationSession) {
            haSipApplicationSession = (HASipApplicationSession)sipApplicationSession;
        }        
        
        //look in replica cache
        ReplicationState replicaState
            = getFromSipApplicationSessionReplicationCache((String)queryState.getId());
        ReplicationState returnState 
            = getBestSipApplicationSession(haSipApplicationSession, replicaState, queryState);          
        clearFromSipApplicationSessionManagerCache((String)queryState.getId());
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindsas:returnState=" + returnState);
        }
        return returnState;
    }
    
    public void processBroadcastloadreceivedsas(ReplicationState queryState) {
        //load is acknowledged safe to remove replica now
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processbroadcastloadreceivedsas:instance: " + instanceName);
            _logger.fine("in " + this.getClass().getName() + ">>processbroadcastloadreceivedsas:id=" + queryState.getId());                        
        }
        if(queryState == null || queryState.getId() == null) {
            return;
        }
        String id = (String)queryState.getId();
        //remove active sip application session if present and not locked
        SipApplicationSessionImpl sipApplicationSession 
            = findSipApplicationSessionFromCacheOnly(id);
        if(sipApplicationSession != null 
                && !sipApplicationSession.isForegroundLocked()) {
            sipApplicationSession.passivate();
        } 
        //only safe to remove replica if we are not the replica partner
        //of the sending instance
        ReplicationHealthChecker healthChecker 
            = ReplicationHealthChecker.getInstance();
        String replicatedFromInstanceName 
            = healthChecker.getReshapeReplicateFromInstanceName();
        if(replicatedFromInstanceName != null && !replicatedFromInstanceName.equalsIgnoreCase(queryState.getInstanceName())) {            
            removeFromSipApplicationSessionReplicationCache(id);           
        }
    }
    
    /**
     * process the loadAdvisory of a Servlet Timer
     * this should trigger a load call for the ServletTimer
     * @param timerState - contains the remove command
     */     
    public void processLoadadvisorysas(ReplicationState replicationState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processLoadadvisoryservlettimer");
        } 
        if(replicationState == null) {
            return;
        }
        //try to load from local cache first
        String id = (String)replicationState.getId();
        HASipApplicationSession sipApplicationSession = (HASipApplicationSession)
                this.findSipApplicationSessionFromCacheOnly(id);
        if (sipApplicationSession != null) {
            //if we already have an active sas - just short-circuit by calling
            //load received
            JxtaReplicationSender sender
                = JxtaReplicationSender.createInstance();
            ReplicationState loadReceivedState = 
                ReplicationState.createBroadcastLoadReceivedState(MODE_SIP, id, this.getApplicationId(), sipApplicationSession.getVersion(), instanceName);
            sender.sendBroadcastQuery(loadReceivedState);
            sipApplicationSession.activateGraph();
        } else {
            //load and activate SipApplicationSession
            try {
                sipApplicationSession = (HASipApplicationSession) swapInSipApplicationSession(id);
                if (sipApplicationSession != null) {
                    sipApplicationSession.activateGraph();
                }
            } catch (IOException ex) {
                _logger.log(Level.WARNING,
                    "unable to load SipApplicationSession:id = " + id,
                    ex);
            }
        }
    }
    
    /**
     * find the best version of SipApplicationSession
     * removing any stale versions and return query result
     * @param activeSipApplicationSession SipApplicationSession 
     *      from active cache
     * @param replicaSipApplicationSession SipApplicationSession 
     *      from replica cache
     * @param queryVersion version requested in query (-1 means 
     *      version unaware
     */     
    private ReplicationState getBestSipApplicationSession(HASipApplicationSession activeSipApplicationSession, 
        ReplicationState replicaSipApplicationSession, ReplicationState queryState) {
        ReplicationState bestResult = null;
        long queryVersion = queryState.getVersion();
        
        //first check for none found in either active or replica caches
        if(replicaSipApplicationSession == null && activeSipApplicationSession == null) {
            //return nack
            return ReplicationState.createQueryResponseFrom(queryState, true);
        }
        
        //next check for artifacts found in both active and replica caches
        if(replicaSipApplicationSession != null && activeSipApplicationSession != null) { 
            //compare and remove the lesser version
            //keeping the higher version as (tentative) best
            if(replicaSipApplicationSession.getVersion() < activeSipApplicationSession.getVersion()) {
                //remove stale replica - work with active
                removeFromSipApplicationSessionReplicationCache((String)replicaSipApplicationSession.getId());
                //create appropriate response from active
                bestResult 
                    = createSipApplicationSessionResponseFrom(activeSipApplicationSession, queryState);
            } else {
                //remove stale active - work with replica
                clearFromSipApplicationSessionManagerCache(activeSipApplicationSession.getId());               
                //create appropriate response from replica
                bestResult 
                    = createSipApplicationSessionResponseFrom(replicaSipApplicationSession, queryState);                                
            }
        } else {
            //either replica or active is null and other is non-null
            //replica is null and active is not null
            if(replicaSipApplicationSession == null) {                
                //create appropriate response from active
                bestResult 
                    = createSipApplicationSessionResponseFrom(activeSipApplicationSession, queryState);                               
            } else {
                //active is null & replica is not null
                //create appropriate response from replica
                bestResult 
                    = createSipApplicationSessionResponseFrom(replicaSipApplicationSession, queryState);                                 
            }
        }
        return bestResult;
    }
    
    private ReplicationState createSipApplicationSessionResponseFrom(ReplicationState replicaSipApplicationSession, 
            ReplicationState queryState) {        
        //create appropriate response from replica
        ReplicationState result = null;
        long queryVersion = queryState.getVersion();        
        if(queryVersion != -1 && replicaSipApplicationSession.getVersion() < queryVersion) {
            //return nack & clear stale replica
            removeFromSipApplicationSessionReplicationCache((String)replicaSipApplicationSession.getId());
            result = ReplicationState.createQueryResponseFrom(queryState, true);
        } else {
            //return real response based on replica
            result = ReplicationState.createQueryResponseFrom(replicaSipApplicationSession); 
        } 
        return result;
    }

    private ReplicationState createSipApplicationSessionResponseFrom(HASipApplicationSession activeSipApplicationSession, 
            ReplicationState queryState) {     
        //create appropriate response from active
        ReplicationState result = null;
        long queryVersion = queryState.getVersion();        
        if(queryVersion != -1 && activeSipApplicationSession.getVersion() < queryVersion) {
            //return nack & clear stale active
            clearFromSipApplicationSessionManagerCache(activeSipApplicationSession.getId());
            result = ReplicationState.createQueryResponseFrom(queryState, true);
        } else {
            //return real response based on active SAS
            try {
                result = createQueryResponseFrom(activeSipApplicationSession);
            } catch (IOException ioe) {
                _logger.warning("Failed load: Unable to serialize " +
                                activeSipApplicationSession);
                // We've been unable to serialize the given active
                // SipApplicationSession.
                // Clear it from the active cache and return a nack instead
                clearFromSipApplicationSessionManagerCache(
                    activeSipApplicationSession.getId());
                result = ReplicationState.createQueryResponseFrom(queryState,
                                                                  true);
            }
        }

        return result;
    }
    
    /**
     * Converts the given SipApplicationSession to a ReplicationState.
     *
     * @param sas The SipApplicationSession to be converted
     *
     * @return The ReplicationState corresponding to the given 
     * SipApplicationSession
     */ 
    private ReplicationState createQueryResponseFrom(
            HASipApplicationSession sas) throws IOException {
        byte[] containerExtraParamState = null;
        SipApplicationSessionExtraParams containerExtraParams 
            = sas.getExtraParameters();
        if(containerExtraParams != null) {
            try {
                containerExtraParamState
                    = ReplicationUtil.getByteArray(containerExtraParams);
            } catch (IOException ex) {
                ; //deliberate no-op
            }
        }        
        return new ReplicationState(
                MODE_SIP,
                sas.getId(),
                getApplicationId(),  
                sas.getVersion(),
                0L,
                0L,
                null,
                null,
                null,
                ReplicationState.RETURN_BROADCAST_MSG_COMMAND,
                ReplicationUtil.getByteArray(sas),    
                null,
                containerExtraParamState);
    }

    protected void clearFromSipApplicationSessionManagerCache(String id) {
        SipApplicationSessionImpl sess 
            = this.findSipApplicationSessionFromCacheOnly(id);
        if(sess != null) {                               
            this.removeSipApplicationSessionFromCache(sess);
        } 
    }    
    
    //SipSession process methods
    
    /**
     * process the save of a SipSession
     * @param sessionState - contains the save command
     */     
    public void processSavesipsession(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processSavess");
        }        
        this.putInSipSessionReplicationCache(sessionState);
    }    
    
    /**
     * process the removal of a Sip Session
     * @param sessionState - contains the remove command
     */     
    public void processRemovesipsession(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processRemovess");
        } 
        if(sessionState == null) {
            return;
        }
        replicatedSipSessions.remove(sessionState.getId());
        removeFromSipSessionReplicationUpdateCache((String)sessionState.getId());
    }
    
    /**
     * process the update of a SipSession
     * @param sessionState - contains the updatelastaccesstime command
     */    
    public void processUpdatelastaccesstimesipsession(ReplicationState sessionState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processUpdatelastaccesstimess");
            _logger.fine("processUpdatelastaccesstimess:version:" + sessionState.getVersion());
        }
        String id = (String)sessionState.getId();
        ReplicationState state = this.getFromSipSessionReplicationCache(id);
        if(state != null) {
            state.setLastAccess(sessionState.getLastAccess());
            state.setVersion(sessionState.getVersion());
            state.setContainerExtraParamsState(sessionState.getContainerExtraParamsState());
            this.putInSipSessionReplicationCache(state);
        } else {
            if(_logger.isLoggable(Level.FINE)) {                
                _logger.fine("processUpdatelastaccesstimess: attempting to update a session not yet stored:id:" + sessionState.getId());
            }
            this.putInSipSessionReplicationUpdateCache(new ReplicationStateUpdate(id, sessionState.getVersion(), sessionState.getLastAccess(), sessionState.getContainerExtraParamsState()) );
        }        
    }

    /**
     * process the broadcastfindsipsession for SipSession
     * @param queryState
     */     
    public ReplicationState processBroadcastfindsipsession(ReplicationState queryState) {
        //complete query and send back response
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindsipsession:instance: " + instanceName);
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindsipsession:id=" + queryState.getId());                        
        } 
        
        //look in active cache
        HASipSession haSipSession = null;
        SipSessionDialogImpl sipSession 
            = findSipSessionFromCacheOnly((String)queryState.getId());
        if(sipSession instanceof HASipSession) {
            haSipSession = (HASipSession)sipSession;
        }        
        
        //look in replica cache
        ReplicationState replicaState
            = getFromSipSessionReplicationCache((String)queryState.getId());
        ReplicationState returnState 
            = getBestSipSession(haSipSession, replicaState, queryState);          
        clearFromSipSessionManagerCache((String)queryState.getId());
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindsipsession:returnState=" + returnState);
        }
        return returnState;
    }
    
    public void processBroadcastloadreceivedsipsession(ReplicationState queryState) {
        //load is acknowledged safe to remove replica now
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processbroadcastloadreceivedsipsession:instance: " + instanceName);
            _logger.fine("in " + this.getClass().getName() + ">>processbroadcastloadreceivedsipsession:id=" + queryState.getId());                        
        }
        if(queryState == null || queryState.getId() == null) {
            return;
        }
        String id = (String)queryState.getId();
        //remove active sip session if present and is not locked
        SipSessionDialogImpl sipSession = findSipSessionFromCacheOnly(id);
        if(sipSession != null 
                && !sipSession.isForegroundLocked()) {
            sipSession.notifySessionWillPassivate();
            this.removeSipSessionFromCache(sipSession);
        } 
        //only safe to remove replica if we are not the replica partner
        //of the sending instance
        ReplicationHealthChecker healthChecker 
            = ReplicationHealthChecker.getInstance();
        String replicatedFromInstanceName 
            = healthChecker.getReshapeReplicateFromInstanceName();
        if(replicatedFromInstanceName != null && !replicatedFromInstanceName.equalsIgnoreCase(queryState.getInstanceName())) {           
            removeFromSipSessionReplicationCache(id);           
        }
    }    

    /**
     * find the best version of SipSession
     * removing any stale versions and return query result
     * @param activeSipSession SipSession from active cache
     * @param replicaSas replica SipSession from replica cache
     * @param queryVersion version requested in query (-1 means 
     *      version unaware
     */     
    private ReplicationState getBestSipSession(HASipSession activeSipSession, 
        ReplicationState replicaSipSession, ReplicationState queryState) {
        ReplicationState bestResult = null;
        long queryVersion = queryState.getVersion();
        
        //first check for none found in either active or replica caches
        if(replicaSipSession == null && activeSipSession == null) {
            //return nack
            return ReplicationState.createQueryResponseFrom(queryState, true);
        }
        
        //next check for artifacts found in both active and replica caches
        if(replicaSipSession != null && activeSipSession != null) { 
            //compare and remove the lesser version
            //keeping the higher version as (tentative) best
            if(replicaSipSession.getVersion() < activeSipSession.getVersion()) {
                //remove stale replica - work with active
                removeFromSipSessionReplicationCache((String)replicaSipSession.getId());
                //create appropriate response from active
                bestResult 
                    = createSipSessionResponseFrom(activeSipSession, queryState);
            } else {
                //remove stale active - work with replica
                this.clearFromSipSessionManagerCache(activeSipSession.getId());                
                //create appropriate response from replica
                bestResult 
                    = createSipSessionResponseFrom(replicaSipSession, queryState);                                
            }
        } else {
            //either replica or active is null and other is non-null
            //replica is null and active is not null
            if(replicaSipSession == null) {                
                //create appropriate response from active
                bestResult 
                    = createSipSessionResponseFrom(activeSipSession, queryState);                               
            } else {
                //active is null & replica is not null
                //create appropriate response from replica
                bestResult 
                    = createSipSessionResponseFrom(replicaSipSession, queryState);                                 
            }
        }
        return bestResult;
    }
    
    private ReplicationState createSipSessionResponseFrom(ReplicationState replicaSipSession, 
            ReplicationState queryState) {        
        //create appropriate response from replica
        ReplicationState result = null;
        long queryVersion = queryState.getVersion();        
        if(queryVersion != -1 && replicaSipSession.getVersion() < queryVersion) {
            //return nack & clear stale replica
            removeFromSipSessionReplicationCache((String)replicaSipSession.getId());
            result = ReplicationState.createQueryResponseFrom(queryState, true);
        } else {
            //return real response based on replica
            result = ReplicationState.createQueryResponseFrom(replicaSipSession); 
        } 
        return result;
    }

    private ReplicationState createSipSessionResponseFrom(HASipSession activeSipSession, 
            ReplicationState queryState) {     
        //create appropriate response from active
        ReplicationState result = null;
        long queryVersion = queryState.getVersion();        
        if(queryVersion != -1 && activeSipSession.getVersion() < queryVersion) {
            //return nack & clear stale active
            clearFromSipSessionManagerCache(activeSipSession.getId());
            result = ReplicationState.createQueryResponseFrom(queryState, true);
        } else {
            //return real response based on active SipSession
            try {
                result = createQueryResponseFrom(activeSipSession);
            } catch (IOException ioe) {
                _logger.warning("Failed load: Unable to serialize " +
                                activeSipSession);
                // We've been unable to serialize the given active SipSession.
                // Clear it from the active cache and return a nack instead
                clearFromSipSessionManagerCache(activeSipSession.getId());
                result = ReplicationState.createQueryResponseFrom(queryState,
                                                                  true);
            }
        }
        return result;
    }
    
    /**
     * Converts the given SipSession to a ReplicationState.
     *
     * @param ss The SipSession to be converted
     *
     * @return The ReplicationState corresponding to the given SipSession
     */ 
    private ReplicationState createQueryResponseFrom(HASipSession ss)
            throws IOException {
        byte[] containerExtraParamState = null;
        SipSessionExtraParams containerExtraParams 
            = ss.getExtraParameters();
        if(containerExtraParams != null) {
            try {
                containerExtraParamState
                    = ReplicationUtil.getByteArray(containerExtraParams);
            } catch (IOException ex) {
                ; //deliberate no-op
            }
        }        
        return new ReplicationState(
                MODE_SIP,
                ss.getId(),
                getApplicationId(),  
                ss.getVersion(),
                0L,
                0L,
                null,
                null,
                null,
                ReplicationState.RETURN_BROADCAST_MSG_COMMAND,
                ReplicationUtil.getByteArray(ss),
                null,
                containerExtraParamState);
    }
    
    protected void clearFromSipSessionManagerCache(String id) {
        SipSessionDialogImpl sess 
            = this.findSipSessionFromCacheOnly(id);
        if(sess != null) {                               
            this.removeSipSessionFromCache(sess);
        } 
    }    
    
    //ServletTimer process methods
    
    /**
     * process the save of a Servlet Timer
     * @param timerState - contains the save command
     */     
    public void processSaveservlettimer(ReplicationState timerState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processSaveservlettimer");
        }        
        this.putInServletTimerReplicationCache(timerState);
    }     
    
    /**
     * process the removal of a Servlet Timer
     * @param timerState - contains the remove command
     */     
    public void processRemoveservlettimer(ReplicationState timerState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processRemoveservlettimer");
        } 
        if(timerState == null) {
            return;
        }
        //  call unregisterServletTimerExtraParam()
        if(activationHelper.hasTimerTask((String)timerState.getId())) {
            activationHelper.unregisterServletTimerExtraParam((String)timerState.getId());
        }        
        replicatedServletTimers.remove(timerState.getId());
        removeFromServletTimerReplicationUpdateCache((String)timerState.getId());
    }
    
    /**
     * process the update of a Servlet Timer
     * @param timerState - contains the updatelastaccesstime command
     */    
    public void processUpdatelastaccesstimeservlettimer(ReplicationState timerState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processUpdatelastaccesstimeservlettimer");
            _logger.fine("processUpdatelastaccesstimeservlettimer:version:" + timerState.getVersion());
        }
        String id = (String)timerState.getId();
        ReplicationState state = this.getFromServletTimerReplicationCache(id);
        if(state != null) {
            state.setLastAccess(timerState.getLastAccess());
            state.setVersion(timerState.getVersion());
            state.setContainerExtraParamsState(timerState.getContainerExtraParamsState());
            this.putInServletTimerReplicationCache(state);
        } else {
            if(_logger.isLoggable(Level.FINE)) {                
                _logger.fine("processUpdatelastaccesstimeservlettimer: attempting to update a session not yet stored:id:" + timerState.getId());
            }
            this.putInServletTimerReplicationUpdateCache(new ReplicationStateUpdate(id, timerState.getVersion(), timerState.getLastAccess(), timerState.getContainerExtraParamsState()) );
        }        
    }
    
    /**
     * process the broadcastfindservlettimer for ServletTimer
     * @param queryState
     */     
    public ReplicationState processBroadcastfindservlettimer(ReplicationState queryState) {
        //complete query and send back response
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindservlettimer:instance: " + instanceName);
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindservlettimer:id=" + queryState.getId());                        
        } 
        
        //look in active cache
        HAServletTimer haServletTimer = null;
        ServletTimerImpl servletTimer 
            = findServletTimerFromCacheOnly((String)queryState.getId());
        if(servletTimer instanceof HAServletTimer) {
            haServletTimer = (HAServletTimer)servletTimer;
        }        
        
        //look in replica cache
        ReplicationState replicaState
            = getFromServletTimerReplicationCache((String)queryState.getId());
        ReplicationState returnState 
            = getBestServletTimer(haServletTimer, replicaState, queryState);          
        clearFromServletTimerManagerCache((String)queryState.getId());
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastfindservlettimer:returnState=" + returnState);
        }
        return returnState;
    }
    
    public void processBroadcastloadreceivedservlettimer(ReplicationState queryState) {
        //load is acknowledged safe to remove replica now
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processbroadcastloadreceivedservlettimer:instance: " + instanceName);
            _logger.fine("in " + this.getClass().getName() + ">>processbroadcastloadreceivedservlettimer:id=" + queryState.getId());                        
        }
        if(queryState == null || queryState.getId() == null) {
            return;
        }
        String id = (String)queryState.getId();
        //  call unregisterServletTimerExtraParam()
        if(activationHelper.hasTimerTask(id)) {
            activationHelper.unregisterServletTimerExtraParam(id);
        }
        //remove active timer if present and not locked
        ServletTimerImpl timer = findServletTimerFromCacheOnly(id);
        if(timer != null 
                && !timer.isForegroundLocked()) {        
            this.removeServletTimerFromCache(timer);
        } 
        //only safe to remove replica if we are not the replica partner
        //of the sending instance
        ReplicationHealthChecker healthChecker 
            = ReplicationHealthChecker.getInstance();
        String replicatedFromInstanceName 
            = healthChecker.getReshapeReplicateFromInstanceName();
        if(replicatedFromInstanceName != null && !replicatedFromInstanceName.equalsIgnoreCase(queryState.getInstanceName())) {           
            removeFromServletTimerReplicationCache(id);           
        }
    }
    
    /**
     * process the loadAdvisory of a Servlet Timer
     * this should trigger a load call for the ServletTimer
     * @param timerState - contains the remove command
     */     
    public void processLoadadvisoryservlettimer(ServletTimerExtraParams extraParams) {
        if(extraParams == null) {
            return;
        }        
        processLoadadvisoryservlettimer(extraParams.getId());
        //Note: processBroadcastloadreceivedservlettimer will eventually
        //  call unregisterServletTimerExtraParam()        
    }

    public void processLoadadvisoryservlettimer(ReplicationState timerState) {
        if(timerState == null) {
            return;
        }        
        processLoadadvisoryservlettimer((String)timerState.getId());
    } 
    
    public void processLoadadvisoryservlettimer(String id) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("IN" + this.getClass().getName() + ">>processLoadadvisoryservlettimer");
        } 
        if(id == null) {
            return;
        }
        //try to load from local cache first
        ServletTimerImpl activeTimer = this.findServletTimerFromCacheOnly(id);
        if(activeTimer != null) {
            //if we already have an active timer - just short-circuit by calling
            //load received
            JxtaReplicationSender sender
                = JxtaReplicationSender.createInstance();
            ReplicationState loadReceivedState = 
                ReplicationState.createBroadcastLoadReceivedState(MODE_SIP, id, this.getApplicationId(), ((HAServletTimer)activeTimer).getVersion(), instanceName);
            sender.sendBroadcastQuery(loadReceivedState);            
        } else {
            //load and activate ServletTimer
            try {
                swapInServletTimer(id);
            } catch (IOException ex) {
                _logger.log(Level.WARNING,
                    "unable to load ServletTimer:id = " + id,
                    ex);
            }
        }
    }    
    
    /**
     * find the best version of ServletTimer
     * removing any stale versions and return query result
     * @param activeServletTimer ServletTimer 
     *      from active cache
     * @param replicaSipApplicationSession ServletTimer replica 
     *      from replica cache
     * @param queryVersion version requested in query (-1 means 
     *      version unaware
     */     
    private ReplicationState getBestServletTimer(HAServletTimer activeServletTimer, 
        ReplicationState replicaServletTimer, ReplicationState queryState) {
        ReplicationState bestResult = null;
        long queryVersion = queryState.getVersion();
        
        //first check for none found in either active or replica caches
        if(replicaServletTimer == null && activeServletTimer == null) {
            //return nack
            return ReplicationState.createQueryResponseFrom(queryState, true);
        }
        
        //next check for artifacts found in both active and replica caches
        if(replicaServletTimer != null && activeServletTimer != null) { 
            //compare and remove the lesser version
            //keeping the higher version as (tentative) best
            if(replicaServletTimer.getVersion() < activeServletTimer.getVersion()) {
                //remove stale replica - work with active
                removeFromServletTimerReplicationCache((String)replicaServletTimer.getId());
                //create appropriate response from active
                bestResult 
                    = createServletTimerResponseFrom(activeServletTimer, queryState);
            } else {
                //remove stale active - work with replica
                clearFromServletTimerManagerCache(activeServletTimer.getId());               
                //create appropriate response from replica
                bestResult 
                    = createServletTimerResponseFrom(replicaServletTimer, queryState);                                
            }
        } else {
            //either replica or active is null and other is non-null
            //replica is null and active is not null
            if(replicaServletTimer == null) {                
                //create appropriate response from active
                bestResult 
                    = createServletTimerResponseFrom(activeServletTimer, queryState);                               
            } else {
                //active is null & replica is not null
                //create appropriate response from replica
                bestResult 
                    = createServletTimerResponseFrom(replicaServletTimer, queryState);                                 
            }
        }
        return bestResult;
    }
    
    private ReplicationState createServletTimerResponseFrom(ReplicationState replicaServletTimer, 
            ReplicationState queryState) {        
        //create appropriate response from replica
        ReplicationState result = null;
        long queryVersion = queryState.getVersion();        
        if(queryVersion != -1 && replicaServletTimer.getVersion() < queryVersion) {
            //return nack & clear stale replica
            removeFromServletTimerReplicationCache((String)replicaServletTimer.getId());
            result = ReplicationState.createQueryResponseFrom(queryState, true);
        } else {
            //return real response based on replica
            result = ReplicationState.createQueryResponseFrom(replicaServletTimer); 
        } 
        return result;
    }

    private ReplicationState createServletTimerResponseFrom(HAServletTimer activeServletTimer, 
            ReplicationState queryState) {     
        //create appropriate response from active
        ReplicationState result = null;
        long queryVersion = queryState.getVersion();        
        if(queryVersion != -1 && activeServletTimer.getVersion() < queryVersion) {
            //return nack & clear stale active
            clearFromServletTimerManagerCache(activeServletTimer.getId());
            result = ReplicationState.createQueryResponseFrom(queryState, true);
        } else {
            //return real response based on active ServletTimer
            try {
                result = createQueryResponseFrom(activeServletTimer);
            } catch (IOException ioe) {
                _logger.warning("Failed load: Unable to serialize " +
                                activeServletTimer);
                // We've been unable to serialize the given active
                // ServletTimer.
                // Clear it from the active cache and return a nack instead
                clearFromServletTimerManagerCache(activeServletTimer.getId());
                result = ReplicationState.createQueryResponseFrom(queryState,
                                                                  true);
            }
        }
        return result;
    }
    
    /**
     * Converts the given ServletTimer to a ReplicationState.
     *
     * @param timer The ServletTimer to be converted
     *
     * @return The ReplicationState corresponding to the given ServletTimer
     */ 
    private ReplicationState createQueryResponseFrom(
            HAServletTimer timer) throws IOException {
        byte[] containerExtraParamState = null;
        ServletTimerExtraParams containerExtraParams 
            = timer.getExtraParameters();
        if(containerExtraParams != null) {
            try {
                containerExtraParamState
                    = ReplicationUtil.getByteArray(containerExtraParams);
            } catch (IOException ex) {
                ; //deliberate no-op
            }
        }
        return new ReplicationState(
                MODE_SIP,
                timer.getId(),
                getApplicationId(),  
                timer.getVersion(),
                0L,
                0L,
                null,
                null,
                null,
                ReplicationState.RETURN_BROADCAST_MSG_COMMAND,
                ReplicationUtil.getByteArray(timer),
                null,
                containerExtraParamState);
    }

    protected void clearFromServletTimerManagerCache(String id) {
        ServletTimerImpl timer 
            = this.findServletTimerFromCacheOnly(id);
        if(timer != null) {                               
            this.removeServletTimerFromCache(timer);
        } 
    }
    
    public void purge(String owningInstanceName, long purgeStartTime) {
        purgeSipApplicationSessionsForOwningInstance(owningInstanceName);
        purgeSipSessionsForOwningInstance(owningInstanceName);
        purgeServletTimersForOwningInstance(owningInstanceName);
    }
    
    void purgeSipApplicationSessionsForOwningInstance(String owningInstanceName) {
        List idsToRemove = new ArrayList();
        Iterator it = replicatedSipApplicationSessions.values();
        while(it.hasNext()) {
            ReplicationState nextState 
                = (ReplicationState)it.next();
            byte[] extraParamsState
                = nextState.getContainerExtraParamsState();
            SipApplicationSessionExtraParams extraParams = null;
            if(extraParamsState != null) {
                try {
                extraParams
                    = (SipApplicationSessionExtraParams)ReplicationState.getObjectValue(extraParamsState);
                } catch (Exception ex) {
                    _logger.log(Level.WARNING,
                        "unable to deserialize SipApplicationSession extraParams for id " + nextState.getId(),
                        ex);
                }
            }
            if(extraParams != null) {
                String nextOwningInstance = extraParams.getCurrentOwnerInstanceName();
                if(nextOwningInstance != null 
                    && nextOwningInstance.equalsIgnoreCase(owningInstanceName)) {
                    idsToRemove.add((String)nextState.getId());
                }
            }
        }
        for(int i=0; i< idsToRemove.size(); i++) {
            removeFromSipApplicationSessionReplicationCache((String)idsToRemove.get(i));
        }
    }
    
    void purgeSipSessionsForOwningInstance(String owningInstanceName) {
        List idsToRemove = new ArrayList();
        Iterator it = replicatedSipSessions.values();
        while(it.hasNext()) {
            ReplicationState nextState 
                = (ReplicationState)it.next();
            byte[] extraParamsState
                = nextState.getContainerExtraParamsState();
            SipSessionExtraParams extraParams = null;
            if(extraParamsState != null) {
                try {
                extraParams
                    = (SipSessionExtraParams)ReplicationState.getObjectValue(extraParamsState);
                } catch (Exception ex) {
                    _logger.log(Level.WARNING,
                        "unable to deserialize SipSession extraParams for id " + nextState.getId(),
                        ex);
                }
            }
            if(extraParams != null) {
                String nextOwningInstance = extraParams.getCurrentOwnerInstanceName();
                if(nextOwningInstance != null 
                    && nextOwningInstance.equalsIgnoreCase(owningInstanceName)) {
                    idsToRemove.add((String)nextState.getId());
                }
            }
        }
        for(int i=0; i< idsToRemove.size(); i++) {
            removeFromSipSessionReplicationCache((String)idsToRemove.get(i));
        }
    }
    
    void purgeServletTimersForOwningInstance(String owningInstanceName) {
        List idsToRemove = new ArrayList();
        Iterator it = replicatedServletTimers.values();
        while(it.hasNext()) {
            ReplicationState nextState 
                = (ReplicationState)it.next();
            byte[] extraParamsState
                = nextState.getContainerExtraParamsState();
            ServletTimerExtraParams extraParams = null;
            if(extraParamsState != null) {
                try {
                extraParams
                    = (ServletTimerExtraParams)ReplicationState.getObjectValue(extraParamsState);
                } catch (Exception ex) {
                    _logger.log(Level.WARNING,
                        "unable to deserialize ServletTimer extraParams for id " + nextState.getId(),
                        ex);
                }
            }
            if(extraParams != null) {
                String nextOwningInstance = extraParams.getCurrentOwnerInstanceName();
                if(nextOwningInstance != null 
                    && nextOwningInstance.equalsIgnoreCase(owningInstanceName)) {
                    idsToRemove.add((String)nextState.getId());
                }
            }
        }
        for(int i=0; i< idsToRemove.size(); i++) {
            removeFromServletTimerReplicationCache((String)idsToRemove.get(i));
        }
    }    
    
    //end receiver side process methods
    
    //start expiration related methods
    
    void processExpiredSipApplicationSessionReplicas() {
        //code assumes that isExpired works for this type of ReplicationState
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processExpiredSipApplicationSessionReplicas");            
        }
        ArrayList expiredReplicas = new ArrayList(30);
        BaseCache replicasCache = this.getReplicatedSipApplicationSessions();
        for(Iterator it = replicasCache.values(); it.hasNext();) {
            ReplicationState nextState = (ReplicationState)it.next();
            if(_logger.isLoggable(Level.FINE)) {
                _logger.fine("in " + this.getClass().getName() + "nextState=" + nextState);            
            } 
            if(isTimeForSipApplicationSessionLoadAdvisory(nextState)) {
                expiredReplicas.add(nextState);
            }
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("processExpiredSipApplicationSessionReplicas:expiredReplicas.size=" + expiredReplicas.size());            
        }        
        for(int i=0; i<expiredReplicas.size(); i++) {
            ReplicationState nextState = 
                (ReplicationState)expiredReplicas.get(i);
            String rightfulOwnerInstanceName 
                = SipApplicationSessionUtil.getActualServerInstance((String)nextState.getId());
            this.sendLoadAdvisorySipApplicationSession((String)nextState.getId(), rightfulOwnerInstanceName);
        }
    }
    
    private SipApplicationSessionExtraParams getSipApplicationSessionExtraParamsFrom(ReplicationState state) {
        if(state == null) {
            return null;
        }
        if(state.getDeserializedExtraParam() != null) {
            return (SipApplicationSessionExtraParams)state.getDeserializedExtraParam();
        } else {
            //deserialize and cache result in state
            byte[] extraParamsState = state.getContainerExtraParamsState();
            SipApplicationSessionExtraParams extraParams = null;
            if(extraParamsState != null) {
                try {
                    extraParams
                        = (SipApplicationSessionExtraParams)ReplicationState.getObjectValue(extraParamsState);
                } catch (Exception ex) {
                    _logger.log(Level.WARNING,
                        "unable to deserialize SipApplicationSession extraParams for id " + state.getId(),
                        ex);
                }
                state.setDeserializedExtraParam(extraParams);
            }            
            return extraParams;
        }
    }    
    
    protected boolean isTimeForSipApplicationSessionLoadAdvisory(ReplicationState state) {
        SipApplicationSessionExtraParams extraParams
            = getSipApplicationSessionExtraParamsFrom(state);        
        if(extraParams == null) {
            return false;
        } else {
            return applyTimingHeuristicTo(extraParams.getExpirationTime());
        }
    }    
    
    void processExpiredSipSessionReplicas() {
        //code assumes that isExpired works for this type of ReplicationState        
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processExpiredSipSessionReplicas");            
        }
        ArrayList expiredReplicas = new ArrayList(30);
        BaseCache replicasCache = this.getReplicatedSipSessions();
        for(Iterator it = replicasCache.values(); it.hasNext();) {
            ReplicationState nextState = (ReplicationState)it.next();
            if(_logger.isLoggable(Level.FINE)) {
                _logger.fine("in " + this.getClass().getName() + "nextState=" + nextState);            
            }            
            if(isExpiredSipSessionReplica(nextState)) {
                expiredReplicas.add(nextState);
            }
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("processExpiredSipSessionReplicas:expiredReplicas.size=" + expiredReplicas.size());            
        }        
        for(int i=0; i<expiredReplicas.size(); i++) {
            ReplicationState nextState = 
                (ReplicationState)expiredReplicas.get(i);
            this.removeFromSipSessionReplicationCache((String)nextState.getId());
        }
    }
    
    private SipSessionExtraParams getSipSessionExtraParamsFrom(ReplicationState state) {
        if(state == null) {
            return null;
        }
        if(state.getDeserializedExtraParam() != null) {
            return (SipSessionExtraParams)state.getDeserializedExtraParam();
        } else {
            //deserialize and cache result in state
            byte[] extraParamsState = state.getContainerExtraParamsState();
            SipSessionExtraParams extraParams = null;
            if(extraParamsState != null) {
                try {
                    extraParams
                        = (SipSessionExtraParams)ReplicationState.getObjectValue(extraParamsState);
                } catch (Exception ex) {
                    _logger.log(Level.WARNING,
                        "unable to deserialize SipSession extraParams for id " + state.getId(),
                        ex);
                }
                state.setDeserializedExtraParam(extraParams);
            }            
            return extraParams;
        }
    }
    
    protected boolean isExpiredSipSessionReplica(ReplicationState state) {
        SipSessionExtraParams extraParams
            = getSipSessionExtraParamsFrom(state);
        if(extraParams == null) {
            return false;
        } else {
            return (extraParams.getExpirationTime() < System.currentTimeMillis());
        }
    }    
    
    void processExpiredServletTimerReplicas() {
        //code assumes that isExpired works for this type of ReplicationState        
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processExpiredServletTimerReplicas");            
        }
        ArrayList expiredReplicas = new ArrayList(30);
        BaseCache replicasCache = this.getReplicatedServletTimers();
        for(Iterator it = replicasCache.values(); it.hasNext();) {
            ReplicationState nextState = (ReplicationState)it.next();
            if(_logger.isLoggable(Level.FINE)) {
                _logger.fine("in " + this.getClass().getName() + "nextState=" + nextState);            
            }            
            if(isTimeForServetTimerLoadAdvisory(nextState)) {
                expiredReplicas.add(nextState);
            }
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("processExpiredServletTimerReplicas:expiredReplicas.size=" + expiredReplicas.size());            
        }        
        for(int i=0; i<expiredReplicas.size(); i++) {
            ReplicationState nextState = (ReplicationState)expiredReplicas.get(i);
            ServletTimerExtraParams extraParams
                = (ServletTimerExtraParams)nextState.getContainerExtraParams();
            if(extraParams != null && extraParams.getParentSasId() != null) {
                String sasId = extraParams.getParentSasId();
                String rightfulOwnerInstanceName 
                    = SipApplicationSessionUtil.getActualServerInstance(sasId);
                this.sendLoadAdvisoryServletTimer((String)nextState.getId(), rightfulOwnerInstanceName);
            }
        }
    }
    
    private ServletTimerExtraParams getServletTimerExtraParamsFrom(ReplicationState state) {
        if(state == null) {
            return null;
        }
        if(state.getDeserializedExtraParam() != null) {
            return (ServletTimerExtraParams)state.getDeserializedExtraParam();
        } else {
            //deserialize and cache result in state
            byte[] extraParamsState = state.getContainerExtraParamsState();
            ServletTimerExtraParams extraParams = null;
            if(extraParamsState != null) {
                try {
                    extraParams
                        = (ServletTimerExtraParams)ReplicationState.getObjectValue(extraParamsState);
                } catch (Exception ex) {
                    _logger.log(Level.WARNING,
                        "unable to deserialize ServletTimer extraParams for id " + state.getId(),
                        ex);
                }
                state.setDeserializedExtraParam(extraParams);
            }            
            return extraParams;
        }
    }     
    
    protected boolean isTimeForServetTimerLoadAdvisory(ReplicationState state) {
        ServletTimerExtraParams extraParams
            = getServletTimerExtraParamsFrom(state);        
        if(extraParams == null) {
            return false;
        } else {
            return applyTimingHeuristicTo(extraParams.getExpirationTime());
        }
    }
    
    private boolean applyTimingHeuristicTo(long timerExpirationTime) {
        return activationHelper.calculateServletTimerTaskScheduleTime(timerExpirationTime) == 0L;
    }
    
    //end expiration related methods

    public SessionFactory getSessionFactory() {
        return sessionFactory;
    }    

    public void setSessionFactory(SessionFactory sessionFactory) {
        this.sessionFactory = sessionFactory;
    }
    
    public StorePool getSipApplicationSessionStorePool() {
        return sipApplicationSessionStorePool;
    }
    
    public void setSipApplicationSessionStorePool(StorePool sipApplicationSessionStorePool) {
        this.sipApplicationSessionStorePool = sipApplicationSessionStorePool;
    }
    
    public StorePool getSipSessionStorePool() {
        return sipSessionStorePool;
    }
    
    public void setSipSessionStorePool(StorePool sipSessionStorePool) {
        this.sipSessionStorePool = sipSessionStorePool;
    }           

    public StorePool getServletTimerStorePool() {
        return servletTimerStorePool;
    }
    
    public void setServletTimerStorePool(StorePool servletTimerStorePool) {
        this.servletTimerStorePool = servletTimerStorePool;
    }           
    
    /**
     * Creates a new SipSession.
     *
     * @param set
     * @param to
     * @param appSession
     * @param handler
     * @type type
     *
     * @return The new SipSession
     */
    protected SipSessionDialogImpl createNewSipSession(
            DialogSet set, Address to, SipApplicationSessionImpl appSession,
            String handler, Type type) {

        return getSessionFactory().createSipSession(this, set, to, appSession,
                                                    handler, type);
    }


    /**
     * Creates a new SipApplicationSession.
     *
     * @param id The id of the new SipApplicationSession
     *
     * @return The new SipApplicationSession
     */
    protected SipApplicationSessionImpl createNewSipApplicationSession(String id) {

        return getSessionFactory().createSipApplicationSession(this, id);
    }


    /**
     * Creates a new ServletTimer.
     *
     * @return The new ServletTimer
     */
    protected ServletTimerImpl createNewServletTimer(
            SipApplicationSessionImpl sas, Serializable info, long delay,
            TimerListener listener) {
        return new HAServletTimer(this, sas, info, delay, listener);
    }


    /**
     * Creates a new ServletTimer.
     *
     * @return The new ServletTimer
     */
    protected ServletTimerImpl createNewServletTimer(
            SipApplicationSessionImpl sas, Serializable info, long delay,
            boolean fixedDelay, long period, TimerListener listener) {
        return new HAServletTimer(this, sas, info, delay, fixedDelay,
                                  period, listener);
    }


    /**
     * Gets the persistence type of this session manager.
     */
    public String getPersistenceType() {
        return EEPersistenceTypeResolver.REPLICATED_TYPE;
    }
    
    //begin send-side methods
    
    /**
     * Persists the given SipApplicationSession.
     *
     * @param sas The SipApplicationSession to be persisted
     */
    public void saveSipApplicationSession(SipApplicationSessionImpl sas)
            throws IOException {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "saveSipApplicationSession");
        }        
        SipApplicationSessionStoreImpl store = null;
        try {
            store = getSipApplicationSessionStore();            
            store.save(sas);
        }
        finally {
            this.putSipApplicationSessionStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "saveSipApplicationSession");
        }
    } 
    
    /**
     * Removes the given SipApplicationSession from both the active cache
     * and the persistent session store of this session manager,
     *
     * @param sas The SipApplicationSession to be removed
     */
    public void removeSipApplicationSession(SipApplicationSessionImpl sas) {
        
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "removeSipApplicationSession");
        } 
        super.removeSipApplicationSessionFromCache(sas);
        SipApplicationSessionStoreImpl store = null;
        try {
            store = getSipApplicationSessionStore();            
            store.remove(sas.getId());
        } catch (IOException ex) {
            _logger.log(
                Level.WARNING,
                "unable to remove SipApplicationSession:id = " + sas.getId(),
                ex);
        } finally {
            this.putSipApplicationSessionStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "removeSipApplicationSession");
        }
    }
    
    /**
     * send a load advisory message to instance to trigger
     * it to load a given id
     *
     * @param id The id of the SipApplicationSession to be loaded
     * @param instanceName the instance to be sent the load advisory
     */
    public void sendLoadAdvisorySipApplicationSession(String id, String instanceName) {

        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "sendLoadAdvisorySipApplicationSession");
        }        
        SipApplicationSessionStoreImpl store = null;
        try {
            store = getSipApplicationSessionStore();            
            store.sendLoadAdvisory(id, instanceName);
        } catch (IOException ex) {
            _logger.log(
                Level.WARNING,
                "unable to send load advisory to " + instanceName + " for SipApplicationSession:id = " + id,
                ex);
        } finally {
            this.putSipApplicationSessionStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "sendLoadAdvisorySipApplicationSession");
        }
    }       
    
    /**
     * Look for a SipApplicationSession in the Store and, if found, restore
     * it in the Manager's list of active sessions if appropriate.
     * The session will be removed from the Store after swapping
     * in, but will not be added to the active session list if it
     * is invalid or past its expiration.
     * @param id the id of the SAS
     */
    protected SipApplicationSessionImpl swapInSipApplicationSession(String id) throws IOException {
        return swapInSipApplicationSession(id, null);
    }    
    
    /**
     * Look for a SipApplicationSession in the Store and, if found, restore
     * it in the Manager's list of active sessions if appropriate.
     * The session will be removed from the Store after swapping
     * in, but will not be added to the active session list if it
     * is invalid or past its expiration.
     * @param id the id of the SAS
     * @param version - the version of the SAS
     */
    protected SipApplicationSessionImpl swapInSipApplicationSession(String id, String version) throws IOException {
        SipApplicationSessionStoreImpl store = null;
        try {
            store = this.getSipApplicationSessionStore();
            if (store == null)
                return null;

            SipApplicationSessionImpl session = null;
            try {
                if (SecurityUtil.isPackageProtectionEnabled()){
                    try{
                        session = (SipApplicationSessionImpl) AccessController.doPrivileged(new PrivilegedStoreLoadSipApplicationSession(id, version, store));
                    }catch(PrivilegedActionException ex){
                        Exception exception = ex.getException();
                        if (exception instanceof IOException){
                            throw (IOException)exception;
                        } else if (exception instanceof ClassNotFoundException) {
                            throw (ClassNotFoundException)exception;
                        }
                    }
                } else {
                    if (version != null) {
                         session = store.load(id, version);
                    } else {
                         session = store.load(id);
                    }
                }   
            } catch (ClassNotFoundException e) {
                IOException ex1 = 
                    (IOException) new IOException("Error during swapInSipApplicationSession: " + e.getMessage()).initCause(e);
                throw ex1;                 
            }

            if (session == null) {
                return (null);
            }

            boolean isActivated = session.activate();
            if (!isActivated) {
                // FIXME: Should we mark the object as invalid and return it,
                // instead of returning null?
                return null;
            }

            return session;
        } finally {
            this.putSipApplicationSessionStore(store);
        }
    }
    
    /**
     * Persists the given SipSession.
     *
     * @param sipSession The SipSession to be persisted
     */
    public void saveSipSession(SipSessionDialogImpl sipSession)
            throws IOException {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "saveSipSession");
        }        
        SipSessionStoreImpl store = null;
        try {
            store = getSipSessionStore();            
            store.save(sipSession); 
        }
        finally {
            this.putSipSessionStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "saveSipSession");
        }
    }
    
    /**
     * Removes the given SipSession from both the active cache and the
     * persistent session store of this session manager,
     *
     * @param sipSession The SipSession to be removed
     */
    public void removeSipSession(SipSessionDialogImpl sipSession) {
        
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "removeSipSession");
        }
        removeSipSessionFromCache(sipSession);
        SipSessionStoreImpl store = null;
        try {
            store = getSipSessionStore();            
            store.remove(sipSession.getId()); 
        } catch (IOException ex) {
            _logger.log(
                Level.WARNING,
                "unable to remove SipSession:id = " + sipSession.getId(),
                ex);
        } finally {
            this.putSipSessionStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "removeSipSession");
        }
    } 
    
    /**
     * Look for a SipSession in the Store and, if found, restore
     * it in the Manager's list of active sessions if appropriate.
     * The session will be removed from the Store after swapping
     * in, but will not be added to the active session list if it
     * is invalid or past its expiration.
     * @param id the id of the SipSession
     */
    protected SipSessionDialogImpl swapInSipSession(String id) throws IOException {
        return swapInSipSession(id, null);
    }     
    
    /**
     * Look for a SipSession in the Store and, if found, restore
     * it in the Manager's list of active sessions if appropriate.
     * The session will be removed from the Store after swapping
     * in, but will not be added to the active session list if it
     * is invalid or past its expiration.
     * @param id the id of the SipSession
     * @param version - the version of the SipSession
     */
    protected SipSessionDialogImpl swapInSipSession(String id, String version) throws IOException {
        SipSessionStoreImpl store = null;
        try {
            store = this.getSipSessionStore();
            if (store == null)
                return null;

            SipSessionDialogImpl session = null;
            try {
                if (SecurityUtil.isPackageProtectionEnabled()){
                    try{
                        session = (SipSessionDialogImpl) AccessController.doPrivileged(new PrivilegedStoreLoadSipSession(id, version, store));
                    }catch(PrivilegedActionException ex){
                        Exception exception = ex.getException();
                        if (exception instanceof IOException){
                            throw (IOException)exception;
                        } else if (exception instanceof ClassNotFoundException) {
                            throw (ClassNotFoundException)exception;
                        }
                    }
                } else {
                    if (version != null) {
                         session = store.load(id, version);
                    } else {
                         session = store.load(id);
                    }
                }   
            } catch (ClassNotFoundException e) {
                IOException ex1 = 
                    (IOException) new IOException("Error during swapInSipSession: " + e.getMessage()).initCause(e);
                throw ex1; 
            }

            if (session == null)
                return (null);

            if (!session.isValid()) {
                if(_logger.isLoggable(Level.INFO)) {
                    _logger.log(Level.INFO, "session swapped in is invalid or expired");
                }
                return (null);
            }
            // make sure the listeners know about it.
            //FIXME invoke listeners
            addSipSession(session);
            return (session);
        } finally {
            this.putSipSessionStore(store);
        }
    }    
    
    /**
     * Persists the given ServletTimer.
     *
     * @param servletTimer The ServletTimer to be persisted
     */
    public void saveServletTimer(ServletTimerImpl servletTimer)
            throws IOException {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "saveServletTimer");
        }        
        ServletTimerStoreImpl store = null;
        try {
            store = getServletTimerStore();            
            store.save(servletTimer);
        }
        finally {
            this.putServletTimerStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "saveServletTimer");
        }       
    }
    
    /**
     * Removes the given ServletTimer from both the active cache and the
     * persistent session store of this session manager,
     *
     * @param servletTimer The ServletTimer to be removed
     */
    public void removeServletTimer(ServletTimerImpl servletTimer) {

        removeServletTimerFromCache(servletTimer);

        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "removeServletTimer");
        }        
        ServletTimerStoreImpl store = null;
        try {
            store = getServletTimerStore();            
            store.remove(servletTimer.getId());
        } catch (IOException ex) {
            _logger.log(Level.WARNING,
                "unable to remove ServletTimer:id = " + servletTimer.getId(),
                ex);
        } finally {
            this.putServletTimerStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "removeServletTimer");
        }
    }
    
    /**
     * send a load advisory message to instance to trigger
     * it to load a given id
     *
     * @param id The id of the ServletTimer to be loaded
     * @param instanceName the instance to be sent the load advisory
     */
    public void sendLoadAdvisoryServletTimer(String id, String instanceName) {

        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "sendLoadAdvisoryServletTimer");
        }        
        ServletTimerStoreImpl store = null;
        try {
            store = getServletTimerStore();            
            store.sendLoadAdvisory(id, instanceName);
        } catch (IOException ex) {
            _logger.log(Level.WARNING,
                "unable to send load advisory to " + instanceName + " for ServletTimer:id = " + id,
                ex);
        } finally {
            this.putServletTimerStore(store);
        }                 
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "sendLoadAdvisoryServletTimer");
        }
    }   
    
    /**
     * Look for a ServletTimer in the Store and, if found, restore
     * it in the Manager's list of active sessions if appropriate.
     * The session will be removed from the Store after swapping
     * in, but will not be added to the active session list if it
     * is invalid or past its expiration.
     * @param id the id of the ServletTimer
     */
    protected ServletTimerImpl swapInServletTimer(String id) throws IOException {
        return swapInServletTimer(id, null);
    }     
    
    /**
     * Look for a ServletTimer in the Store and, if found, restore
     * it in the Manager's list of active sessions if appropriate.
     * The session will be removed from the Store after swapping
     * in, but will not be added to the active session list if it
     * is invalid or past its expiration.
     * @param id the id of the ServletTimer
     * @param version - the version of the ServletTimer
     */
    protected ServletTimerImpl swapInServletTimer(String id, String version) throws IOException {
        ServletTimerStoreImpl store = null;
        try {
            store = this.getServletTimerStore();
            if (store == null)
                return null;

            ServletTimerImpl timer = null;
            try {
                if (SecurityUtil.isPackageProtectionEnabled()){
                    try{
                        timer = (ServletTimerImpl) AccessController.doPrivileged(new PrivilegedStoreLoadServletTimer(id, version, store));
                    }catch(PrivilegedActionException ex){
                        Exception exception = ex.getException();
                        if (exception instanceof IOException){
                            throw (IOException)exception;
                        } else if (exception instanceof ClassNotFoundException) {
                            throw (ClassNotFoundException)exception;
                        }
                    }
                } else {
                    if (version != null) {
                         timer = store.load(id, version);
                    } else {
                         timer = store.load(id);
                    }
                }   
            } catch (ClassNotFoundException e) {
                 IOException ex1 = 
                    (IOException) new IOException("Error during swapInServletTimer: " + e.getMessage()).initCause(e);
                throw ex1; 
            }

            if (timer == null) {
                return (null);
            }

            boolean isActivated = timer.activate();
            if (!isActivated) {
                // FIXME: Should we mark the object as invalid and return it,
                // instead of returning null?
                return null;
            }

            return timer;
        } finally {
            this.putServletTimerStore(store);
        }
    }     
    
    /** Returns a store from the pool 
     * This method intializes the store with right parameters
     * @return returns SipApplicationSessionStoreImpl
     */
    public SipApplicationSessionStoreImpl getSipApplicationSessionStore() {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "getSipApplicationSessionStore");
        }
        SipApplicationSessionStoreImpl store = null;
        try {
            store = (SipApplicationSessionStoreImpl) sipApplicationSessionStorePool.take();
            store.setSipSessionManager(this);
            if(_logger.isLoggable(Level.FINEST)) {
                _logger.log(Level.FINEST,
                    "SipTransactionPersistentManager.getSipApplicationSessionStore returning   " + store);
            }
            return store;
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "getSipApplicationSessionStore", store);
        }
        return store;
    } 
    
    /** 
     *  Returns (puts) a store back to the pool
     */
    private void putSipApplicationSessionStore(StorePoolElement store) {    
        ( (SipApplicationSessionStoreImpl) store).setSipSessionManager(null);
        if (store != null) {
            try {
                if(sipApplicationSessionStorePool != null) {
                    sipApplicationSessionStorePool.put(store);
                }
            }
            catch (InterruptedException ex1) {
                ex1.printStackTrace();
            }
        }
    }
    
    /** Returns the singleton store 
     * This method intializes the store with right parameters
     * @return returns SipApplicationSessionStoreImpl
     */
    public SipApplicationSessionStoreImpl getSingletonSipApplicationSessionStore() { 
        return (SipApplicationSessionStoreImpl)super.getSipApplicationSessionStore(); 
    }     
    
    /** Returns a store from the pool.
     * This method intializes the store with right parameters
     * @return returns SipSessionStoreImpl
     */
    public SipSessionStoreImpl getSipSessionStore() {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "getSipSessionStore");
        }
        SipSessionStoreImpl store = null;
        try {
            store = (SipSessionStoreImpl) sipSessionStorePool.take();
            store.setSipSessionManager(this);
            if(_logger.isLoggable(Level.FINEST)) {
                _logger.log(Level.FINEST,
                    "SipTransactionPersistentManager.getSipSessionStore returning   " + store);
            }
            return store;
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "getSipSessionStore", store);
        }
        return store;
    }
    
    /** 
     *  Returns (puts) a store back to the pool
     */
    private void putSipSessionStore(StorePoolElement store) {    
        ( (SipSessionStoreImpl) store).setSipSessionManager(null);
        if (store != null) {
            try {
                if(sipSessionStorePool != null) {
                    sipSessionStorePool.put(store);
                }
            }
            catch (InterruptedException ex1) {
                ex1.printStackTrace();
            }
        }
    }
    
    /** Returns the singleton store 
     * This method intializes the store with right parameters
     * @return returns SipSessionStoreImpl
     */
    public SipSessionStoreImpl getSingletonSipSessionStore() { 
        return (SipSessionStoreImpl)super.getSipSessionStore();
    }    
    
    /** Returns a store from the pool 
     * This method intializes the store with right parameters
     * @return returns ServletTimerStoreImpl
     */
    public ServletTimerStoreImpl getServletTimerStore() {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.entering("SipTransactionPersistentManager", "getServletTimerStore");
        }
        ServletTimerStoreImpl store = null;
        try {
            store = (ServletTimerStoreImpl) servletTimerStorePool.take();
            store.setSipSessionManager(this);
            if(_logger.isLoggable(Level.FINEST)) {
                _logger.log(Level.FINEST,
                    "SipTransactionPersistentManager.getServletTimerStore returning   " + store);
            }
            return store;
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        if(_logger.isLoggable(Level.FINE)) {
            _logger.exiting("SipTransactionPersistentManager", "getServletTimerStore", store);
        }
        return store;
    }
    
    /** 
     *  Returns (puts) a store back to the pool
     */
    private void putServletTimerStore(StorePoolElement store) {    
        ( (ServletTimerStoreImpl) store).setSipSessionManager(null);
        if (store != null) {
            try {
                if(servletTimerStorePool != null) {
                    servletTimerStorePool.put(store);
                }
            }
            catch (InterruptedException ex1) {
                ex1.printStackTrace();
            }
        }
    } 
    
    /** Returns the singleton store 
     * This method intializes the store with right parameters
     * @return returns ServletTimerStoreImpl
     */
    public ServletTimerStoreImpl getSingletonServletTimerStore() { 
        return (ServletTimerStoreImpl)super.getServletTimerStore();
    }     
    
    //end send-side methods
    
    //begin implement ReplicationManager
    
    private boolean isBroadcastState(ReplicationState state) {
        return(broadcastMethods.contains(state.getCommand()));
    }
                
    public void processMessage(ReplicationState message) {

        //handle broadcast methods
        if(isBroadcastState(message)) {
            processBroadcastMessage(message);
            return;
        }
        
        //handle non-void methods
        ReplicationStateQueryResponse queryResult = null;
        //do process non-void message (and cannot be response either)
        if(!message.isResponseState() && !message.isVoidMethodReturnState()) {
            //do non-void processing including sending response
            queryResult = this.doProcessQueryMessage(message);
            ReplicationState qResponse = queryResult.getState();
            if(qResponse != null) {
                //sourceInstanceName is preserved in the response
                ReplicationState response = 
                    ReplicationState.createResponseFrom(qResponse);
                if(_logger.isLoggable(Level.FINE)) {
                    _logger.fine("RepMgrBase:responseState=" + response);
                }                
                this.doSendResponse(response);
            }
            return;
        }
        //end do process non-void message        

        boolean isResponse = this.doProcessMessage(message);
        
    }
    
    //return true if message is processResponse
    public boolean doProcessMessage(ReplicationState message) {
        boolean result = false;
        String methodName = ReplicationUtil.getProcessMethodName(message);
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>doProcessMessageName=" + methodName);
        }
        try {
            Class myClass = this.getClass();
            myClass.getMethod(
                methodName,
                    new Class[]{ message.getClass() }).invoke(
                        this, new Object[]{ message });            
        } catch (Throwable t) {
           _logger.log(
               Level.SEVERE,
               "Unable to process replication message with methodName=" +
               methodName, t);
        } 
        if(methodName.equals("processResponse")) {
            result = true;
        }
        return result;
    }    
    
    /**
    * send the response
    *
    * @param sessionState 
    *   The replication state response
    */    
    public void doSendResponse(ReplicationState sessionState) {
               
        JxtaReplicationReceiver replicationReceiver = 
                JxtaReplicationReceiver.createInstance();
        ReplicationState resultState = 
                replicationReceiver.sendReplicationStateResponse(sessionState); 
    }        
    
    public void processBroadcastMessage(ReplicationState message) {
        ReplicationStateQueryResponse response = this.doProcessQueryMessage(message);
        boolean isResponse = response.isResponse();
        ReplicationState responseState = response.getState();
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("processBroadcastMessage:after doProcessQueryMessage:response=" + isResponse);
            _logger.fine("processBroadcastMessage:after doProcessQueryMessage:responseState=" + responseState);
            _logger.fine("processBroadcastMessage:after doProcessQueryMessage:responseStateTrunk=" + responseState.getTrunkState());
            _logger.fine("processBroadcastMessage:after doProcessQueryMessage:responseStateAttr=" + responseState.getState());
            _logger.fine("processBroadcastMessage:after doProcessQueryMessage:responseStateVer=" + responseState.getVersion());
        }

        //don't send a response to a response
        if(!isResponse) {
            //point-to-point response back to sender
            //System.out.println("processBroadcastMessage - need to send back result");
            //doSendQueryResponse(responseState, instanceName);
            doSendQueryResponse(responseState, message.getInstanceName());
        }
    }
    
    /**
    * send the response
    *
    * @param sessionState 
    *   The replication state response
    * @param instanceName  the name of the target instance
    */    
    public void doSendQueryResponse(ReplicationState sessionState, String instanceName) {
        JxtaReplicationSender replicationSender = 
                JxtaReplicationSender.createInstance();
        ReplicationState resultState = 
                replicationSender.sendReplicationStateQueryResponse(sessionState, instanceName); 
    }        
    
    //return true if message is processQueryResponse
    public ReplicationStateQueryResponse doProcessQueryMessage(ReplicationState message) {
        ReplicationState resultState = null;
        String methodName = ReplicationUtil.getProcessMethodName(message);
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>doProcessQueryMessage:methodName=" + methodName);
            _logger.fine("in " + this.getClass().getName() + ">>doProcessQueryMessage:thisInstance=" + instanceName + "SASEreturnInstance=" + message.getInstanceName() );
        }         
        try {
            Class myClass = this.getClass();
            resultState = (ReplicationState) myClass.getMethod(
                methodName,
                    new Class[]{ message.getClass() }).invoke(
                        this, new Object[]{ message });            
        } catch (Throwable t) {
           _logger.log(
               Level.SEVERE,
               "Unable to process replication message with methodName=" +
               methodName, t);
        } 
        boolean isResponse = methodName.equals("processBroadcastresponse");
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>doProcessQueryMessage:resultState=" + resultState);
        }        
        //System.out.println("in " + this.getClass().getName() + ">>doProcessQueryMessage:resultState=" + resultState);
        return new ReplicationStateQueryResponse(resultState, isResponse);
    }
    
    public void processQueryMessage(ReplicationState message, String returnInstance) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processQueryMessage:returnInstance= " + returnInstance);
        }         
        ReplicationStateQueryResponse response = this.doProcessQueryMessage(message);
        boolean isResponse = response.isResponse();
        ReplicationState responseState = response.getState();
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("processQueryMessage:after doProcessQueryMessage:response=" + isResponse);
            _logger.fine("processQueryMessage:after doProcessQueryMessage:responseState=" + responseState);            
        }        
        //don't send a response to a response
        if(!isResponse && responseState != null) {
            //point-to-point response back to sender
            if(_logger.isLoggable(Level.FINE)) {
                _logger.fine("processQueryMessage - need to send back result to " + returnInstance);            
            }             
            doSendQueryResponse(responseState, returnInstance);
        }
    }    
    
    public ReplicationState processBroadcastresponse(ReplicationState queryResponseState) {
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("in " + this.getClass().getName() + ">>processBroadcastresponse:queryResponseState=" + queryResponseState);            
        }
        ReplicationResponseRepository.putFederatedEntry(queryResponseState);
        return queryResponseState;
    }

    public void repair(long repairStartTime) {
       if (!isRepairDuringFailure()) {
           return;
       }        
       repairServletTimers(repairStartTime);
       repairSipSessions(repairStartTime);
       repairSipApplicationSessions(repairStartTime);
    }     
    
    public void repair(long repairStartTime, boolean checkForStopping) {
       if (!isRepairDuringFailure()) {
           return;
       }         
        //FIXME
    }
    
    public void repairSipApplicationSessions(long repairStartTime) {    
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("SipTransactionPersistentManager>>repairSipApplicationSessions");                       
        }         

        if(ReplicationHealthChecker.isStopping()) {
            return;
        }

        Iterator it = applicationSessions.values().iterator();
        while (it.hasNext()) {
            SipApplicationSessionImpl session = (SipApplicationSessionImpl) it.next();
            //by virtue of being in the cache it is considered to 
            //be not expired
            //but since invalidate is called before removal we must check
            if(session.isValid()
                && (session.getId() != null) 
                && isSipApplicationSessionOlderThan(session, repairStartTime)           
                && session.lockBackground()) { 
                try {
                    ((HASipApplicationSession)session).setReplicated(false);
                    ((HASipApplicationSession)session).setDirty(true);
                    saveSipApplicationSession(session);
                } catch (IOException ex) {
                    _logger.log(Level.WARNING,
                        "during repair unable to save SipApplicationSession:id = " + session.getId(),
                        ex);
                } finally {
                    session.unlockBackground();
                }                               
	    }            
        }
    }
    
    protected boolean isSipApplicationSessionOlderThan(SipApplicationSessionImpl session, long aTime) {
        return (session.getLastAccessedTime() <= aTime);
    } 
    
    public void repairSipSessions(long repairStartTime) {    
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("SipTransactionPersistentManager>>repairSipSessions");                       
        }         

        if(ReplicationHealthChecker.isStopping()) {
            return;
        }

        Iterator it = sipSessions.values().iterator();
        while (it.hasNext()) {
            SipSessionDialogImpl session = (SipSessionDialogImpl) it.next();
            //by virtue of being in the cache it is considered to 
            //be not expired
            //but since invalidate is called before removal we must check            
            if(session.isValid()
                && (session.getId() != null)
                && isSipSessionOlderThan(session, repairStartTime)           
                && session.lockBackground()) { 
                try {
                    ((HASipSession)session).setReplicated(false);
                    ((HASipSession)session).setDirty(true);
                    saveSipSession(session);
                } catch (IOException ex) {
                    _logger.log(Level.WARNING,
                        "during repair unable to save SipSession:id = " + session.getId(),
                        ex);
                } finally {
                    session.unlockBackground();
                }                               
	    }            
        }
    }
    
    protected boolean isSipSessionOlderThan(SipSessionDialogImpl session, long aTime) {
        return (session.getLastAccessedTime() <= aTime);
    } 
    
    public void repairServletTimers(long repairStartTime) {    
        if(_logger.isLoggable(Level.FINE)) {
            _logger.fine("SipTransactionPersistentManager>>repairServletTimers");                       
        }         

        if(ReplicationHealthChecker.isStopping()) {
            return;
        }

        Iterator it = servletTimers.values().iterator();
        while (it.hasNext()) {
            ServletTimerImpl timer = (ServletTimerImpl) it.next();
            //by virtue of being in the cache it is considered to 
            //be not expired           
            if(timer.getId() != null 
                && isServletTimerOlderThan(timer, repairStartTime)
                && timer.lockBackground()) { 
                try {
                    ((HAServletTimer)timer).setReplicated(false);
                    ((HAServletTimer)timer).setDirty(true);
                    saveServletTimer(timer);
                } catch (IOException ex) {
                    _logger.log(Level.WARNING,
                        "during repair unable to save ServletTimer:id = " + timer.getId(),
                        ex);
                } finally {
                    timer.unlockBackground();
                }
	    }            
        }
    }
    
    protected boolean isServletTimerOlderThan(ServletTimerImpl timer, long aTime) {
        return (((HAServletTimer)timer).getLastAccessedTime() <= aTime);
    }      
    
    public void respondToFailure(String failedInstanceName, boolean checkForStopping) {
        //only do timer wakeup if failedInstanceName was replicating
        //to us
        ReplicationHealthChecker healthChecker 
            = ReplicationHealthChecker.getInstance();
        String replicatedFromInstanceName 
            = healthChecker.getReshapeReplicateFromInstanceName(failedInstanceName);
        //when third party SPI implementation in use then only former replica partner
        //responds otherwise all surviving instances each respond
        if(isThirdPartyBackingStoreInUse() && !failedInstanceName.equalsIgnoreCase(replicatedFromInstanceName)) {
            return;
        }

        if(isThirdPartyBackingStoreInUse()) {
            respondToFailureThirdPartySPI(failedInstanceName);
            return;
        }

        Iterator<String> results =
            (Iterator<String>) replicatedServletTimers.keys();
        while (results.hasNext()) {
            String timerId = (String) results.next();
            try {
                ServletTimerExtraParams eParam = deserializeServletTimerExtraParams(timerId);
                if (failedInstanceName.equals(eParam.getCurrentOwnerInstanceName())) {
                    activationHelper.registerServletTimerExtraParam(eParam);
                }
            } catch (Exception ex) {
                _logger.log(Level.WARNING, "Error during deserialization", ex);
            }
        }
        
        //now do SipApplicationSessions
        Iterator<String> sasResults =
            (Iterator<String>) replicatedSipApplicationSessions.keys();
        while (sasResults.hasNext()) {
            String sasId = (String) sasResults.next();
            try {
                SipApplicationSessionExtraParams eParam 
                    = deserializeSipApplicationSessionExtraParams(sasId);
               
                if (failedInstanceName.equals(eParam.getCurrentOwnerInstanceName())) {
                    //FIXME add logic to ActivationHelper
                    //activationHelper.registerSipApplicationSessionExtraParam(eParam);
                }
            } catch (Exception ex) {
                _logger.log(Level.WARNING, "Error during deserialization", ex);
            }
        }        
    }
    
    void respondToFailureThirdPartySPI(String failedInstanceName) {
        BackingStore servletTimerBackingStore = this.getServletTimerBackingStore();
        ServletTimerExtraParams timerExtraParamCriteria 
            = ServletTimerExtraParams.createSearchCriteria(failedInstanceName);        
        Collection<ServletTimerExtraParams> timerColl 
            = servletTimerBackingStore.findByCriteria(timerExtraParamCriteria);
        
        Iterator<ServletTimerExtraParams> timerResults =
            (Iterator<ServletTimerExtraParams>) timerColl.iterator();
        while (timerResults.hasNext()) {
            ServletTimerExtraParams eParam = timerResults.next();
            if (failedInstanceName.equals(eParam.getCurrentOwnerInstanceName())) {
                activationHelper.registerServletTimerExtraParam(eParam);
            }            
        }
        
        BackingStore sipApplicationSessionBackingStore 
            = this.getSipApplicationSessionBackingStore();
        SipApplicationSessionExtraParams sasExtraParamCriteria 
            = SipApplicationSessionExtraParams.createSearchCriteria(failedInstanceName);        
        Collection<SipApplicationSessionExtraParams> sasColl 
            = sipApplicationSessionBackingStore.findByCriteria(sasExtraParamCriteria);
        
        Iterator<SipApplicationSessionExtraParams> sasResults =
            (Iterator<SipApplicationSessionExtraParams>) sasColl.iterator();
        while (sasResults.hasNext()) {
            SipApplicationSessionExtraParams eParam = sasResults.next();
            if (failedInstanceName.equals(eParam.getCurrentOwnerInstanceName())) {
                //FIXME add logic to ActivationHelper
                //activationHelper.registerSipApplicationSessionExtraParam(eParam);
            }            
        }         
    }
    
    boolean isThirdPartyBackingStoreInUse() {
        BackingStore sasBackingStore 
            = getSipApplicationSessionBackingStore();
        return (!(sasBackingStore instanceof JxtaBackingStoreImpl));
    }

    
    /**
     * Gets the name of the instance.on which this session manager resides.
     *
     * @return The name of the instance on which this session manager
     * resides
     */
    String getInstanceName() {
        return instanceName;
    }

    private class ReplicationStateUpdate {

        private String _id = null;
        private long _lastAccess = 0L;
        private long _version = -1L;
        private byte[] _containerExtraParams = null;
        
        String getId() {
            return _id;
        }        
        
        long getLastAccess() {
            return _lastAccess;
        }
        
        long getVersion() {
            return _version;
        }
        
        byte[] getContainerExtraParams() {
            return _containerExtraParams;
        }        
        
        public ReplicationStateUpdate(String id, long version, long lastAccess, byte[] containerExtraParams) {
            _id = id;
            _version = version;
            _lastAccess = lastAccess;
            _containerExtraParams = containerExtraParams;
        }
        
    } 
    
    private class PrivilegedStoreLoadSipApplicationSession
        implements PrivilegedExceptionAction {

        private String id;
        private String version;
        private SipApplicationSessionStoreImpl store;
            
        PrivilegedStoreLoadSipApplicationSession(String id, String version, SipApplicationSessionStoreImpl store) {     
            this.id = id;
            this.version = version;
            this.store = store;
        }

        public Object run() throws Exception{
           return store.load(id, version);
        }                       
    }
    
    private class PrivilegedStoreLoadSipSession
        implements PrivilegedExceptionAction {

        private String id;
        private String version;
        private SipSessionStoreImpl store;
            
        PrivilegedStoreLoadSipSession(String id, String version, SipSessionStoreImpl store) {     
            this.id = id;
            this.version = version;
            this.store = store;
        }

        public Object run() throws Exception{
           return store.load(id, version);
        }                       
    }
    
    private class PrivilegedStoreLoadServletTimer
        implements PrivilegedExceptionAction {

        private String id;
        private String version;
        private ServletTimerStoreImpl store;
            
        PrivilegedStoreLoadServletTimer(String id, String version, ServletTimerStoreImpl store) {     
            this.id = id;
            this.version = version;
            this.store = store;
        }

        public Object run() throws Exception{
           return store.load(id, version);
        }                       
    }    

}
