//
//  ========================================================================
//  Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
//  ------------------------------------------------------------------------
//  All rights reserved. This program and the accompanying materials
//  are made available under the terms of the Eclipse Public License v1.0
//  and Apache License v2.0 which accompanies this distribution.
//
//      The Eclipse Public License is available at
//      http://www.eclipse.org/legal/epl-v10.html
//
//      The Apache License v2.0 is available at
//      http://www.opensource.org/licenses/apache2.0.php
//
//  You may elect to redistribute this code under either of these licenses.
//  ========================================================================
//

package org.eclipse.jetty.session.infinispan;

import java.io.IOException;
import java.io.ObjectStreamException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import javax.servlet.http.HttpServletRequest;

import org.eclipse.jetty.server.handler.ContextHandler;
import org.eclipse.jetty.server.handler.ContextHandler.Context;
import org.eclipse.jetty.server.session.AbstractSession;
import org.eclipse.jetty.server.session.AbstractSessionManager;
import org.eclipse.jetty.server.session.MemSession;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
import org.eclipse.jetty.util.thread.Scheduler;
import org.infinispan.Cache;
import org.infinispan.commons.api.BasicCache;
import org.omg.CORBA._IDLTypeStub;

/**
 * InfinispanSessionManager
 * 
 * The data for a session relevant to a particular context is stored in an Infinispan (clustered) cache:
 * <pre>
 * Key:   is the id of the session + the context path + the vhost for the context 
 * Value: is the data of the session
 * </pre>
 * 
 * The key is necessarily complex because the same session id can be in-use by more than one
 * context. In this case, the contents of the session will strictly be different for each
 * context, although the id will be the same.
 * 
 * Sessions are also kept in local memory when they are used by this session manager. This allows
 * multiple different request threads in the same context to call Request.getSession() and
 * obtain the same object.
 * 
 * This session manager support scavenging, which is only done over the set of sessions in its
 * local memory. This can result in some sessions being "stranded" in the cluster cache if no
 * session manager is currently managing it (eg the node managing the session crashed and it
 * was never requested on another node).
 * 
 */
public class InfinispanSessionManager extends AbstractSessionManager
{
    private  final static Logger LOG = Log.getLogger("org.eclipse.jetty.server.session");
    
    /**
     * Clustered cache of sessions
     */
    private BasicCache<String, Object> _cache;
    
    
    /**
     * Sessions known to this node held in memory
     */
    private ConcurrentHashMap<String, InfinispanSessionManager.Session> _sessions;

    
    /**
     * The length of time a session can be in memory without being checked against
     * the cluster. A value of 0 indicates that the session is never checked against
     * the cluster - the current node is considered to be the master for the session.
     *
     */
    private long _staleIntervalSec = 0;
    
    protected Scheduler.Task _task; //scavenge task
    protected Scheduler _scheduler;
    protected Scavenger _scavenger;
    protected long _scavengeIntervalMs = 1000L * 60 * 10; //10mins
    protected boolean _ownScheduler;
    
    
    /**
     * Scavenger
     *
     */
    protected class Scavenger implements Runnable
    {

        @Override
        public void run()
        {
           try
           {
               scavenge();
           }
           finally
           {
               if (_scheduler != null && _scheduler.isRunning())
                   _task = _scheduler.schedule(this, _scavengeIntervalMs, TimeUnit.MILLISECONDS);
           }
        }
    }
    
    
    /*
     * Every time a Session is put into the cache one of these objects
     * is created to copy the data out of the in-memory session, and 
     * every time an object is read from the cache one of these objects
     * a fresh Session object is created based on the data held by this
     * object.
     */
    public class SerializableSessionData implements Serializable
    {
        /**
         * 
         */
        private static final long serialVersionUID = -7779120106058533486L;
        String clusterId;
        String contextPath;
        String vhost;
        long accessed;
        long lastAccessed;
        long createTime;
        long cookieSetTime;
        String lastNode;
        long expiry;
        long maxInactive;
        Map<String, Object> attributes;

        public SerializableSessionData()
        {

        }

       
       public SerializableSessionData(Session s)
       {
           clusterId = s.getClusterId();
           contextPath = s.getContextPath();
           vhost = s.getVHost();
           accessed = s.getAccessed();
           lastAccessed = s.getLastAccessedTime();
           createTime = s.getCreationTime();
           cookieSetTime = s.getCookieSetTime();
           lastNode = s.getLastNode();
           expiry = s.getExpiry();
           maxInactive = s.getMaxInactiveInterval();
           attributes = s.getAttributeMap(); // TODO pointer, not a copy
       }
        
        private void writeObject(java.io.ObjectOutputStream out) throws IOException
        {  
            out.writeUTF(clusterId); //session id
            out.writeUTF(contextPath); //context path
            out.writeUTF(vhost); //first vhost

            out.writeLong(accessed);//accessTime
            out.writeLong(lastAccessed); //lastAccessTime
            out.writeLong(createTime); //time created
            out.writeLong(cookieSetTime);//time cookie was set
            out.writeUTF(lastNode); //name of last node managing
      
            out.writeLong(expiry); 
            out.writeLong(maxInactive);
            out.writeObject(attributes);
        }
        
        private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException
        {
            clusterId = in.readUTF();
            contextPath = in.readUTF();
            vhost = in.readUTF();
            
            accessed = in.readLong();//accessTime
            lastAccessed = in.readLong(); //lastAccessTime
            createTime = in.readLong(); //time created
            cookieSetTime = in.readLong();//time cookie was set
            lastNode = in.readUTF(); //last managing node
            expiry = in.readLong(); 
            maxInactive = in.readLong();
            attributes = (HashMap<String,Object>)in.readObject();
        }
        
    }
    
 
    /**
     * Session
     *
     * Representation of a session in local memory.
     */
    public class Session extends MemSession
    {
        
        private ReentrantLock _lock = new ReentrantLock();
        
        /**
         * The (canonical) context path for with which this session is associated
         */
        private String _contextPath;
        
        
        /**
         * The time in msec since the epoch at which this session should expire
         */
        private long _expiryTime; 
        
        
        /**
         * Time in msec since the epoch at which this session was last read from cluster
         */
        private long _lastSyncTime;
        
        
        /**
         * The workername of last node known to be managing the session
         */
        private String _lastNode;
        
        
        /**
         * If dirty, session needs to be (re)sent to cluster
         */
        protected boolean _dirty=false;
        
        
        /**
         * Any virtual hosts for the context with which this session is associated
         */
        private String _vhost;

        
        /**
         * Count of how many threads are active in this session
         */
        private AtomicInteger _activeThreads = new AtomicInteger(0);
        
        
        /**
         * A new session.
         * 
         * @param request the request
         */
        protected Session (HttpServletRequest request)
        {
            super(InfinispanSessionManager.this,request);
            long maxInterval = getMaxInactiveInterval();
            _expiryTime = (maxInterval <= 0 ? 0 : (System.currentTimeMillis() + maxInterval*1000L));
            _lastNode = getSessionIdManager().getWorkerName();
           setVHost(InfinispanSessionManager.getVirtualHost(_context));
           setContextPath(InfinispanSessionManager.getContextPath(_context));
           _activeThreads.incrementAndGet(); //access will not be called on a freshly created session so increment here
        }
        
        
        protected Session (SerializableSessionData sd)
        {
            super(InfinispanSessionManager.this, sd.createTime, sd.accessed, sd.clusterId);
            _expiryTime = (sd.maxInactive <= 0 ? 0 : (System.currentTimeMillis() + sd.maxInactive*1000L));
            setLastNode(sd.lastNode);
            setContextPath(sd.contextPath);
            setVHost(sd.vhost);
            addAttributes(sd.attributes);
        }
        
        
        /**
         * A restored session.
         * 
         * @param sessionId the session id
         * @param created time created
         * @param accessed time last accessed
         * @param maxInterval max expiry interval
         */
        protected Session (String sessionId, long created, long accessed, long maxInterval)
        {
            super(InfinispanSessionManager.this, created, accessed, sessionId);
            _expiryTime = (maxInterval <= 0 ? 0 : (System.currentTimeMillis() + maxInterval*1000L));
        }
        
        /** 
         * Called on entry to the session.
         * 
         * @see org.eclipse.jetty.server.session.AbstractSession#access(long)
         */
        @Override
        protected boolean access(long time)
        {
            if (LOG.isDebugEnabled())
                LOG.debug("Access session({}) for context {} on worker {}", getId(), getContextPath(), getSessionIdManager().getWorkerName());
            try
            {

                long now = System.currentTimeMillis();
                //lock so that no other thread can call access or complete until the first one has refreshed the session object if necessary
                _lock.lock();
                //a request thread is entering
                if (_activeThreads.incrementAndGet() == 1)
                {
                    //if the first thread, check that the session in memory is not stale, if we're checking for stale sessions
                    if (getStaleIntervalSec() > 0  && (now - getLastSyncTime()) >= (getStaleIntervalSec() * 1000L))
                    {
                        if (LOG.isDebugEnabled())
                            LOG.debug("Acess session({}) for context {} on worker {} stale session. Reloading.", getId(), getContextPath(), getSessionIdManager().getWorkerName());
                        refresh();
                    }
                }
            }
            catch (Exception e)
            {
                LOG.warn(e);
            }
            finally
            {            
                _lock.unlock();
            }

            if (super.access(time))
            {
                int maxInterval=getMaxInactiveInterval();
                _expiryTime = (maxInterval <= 0 ? 0 : (time + maxInterval*1000L));
                return true;
            }
            return false;
        }


        /**
         * Exit from session
         * @see org.eclipse.jetty.server.session.AbstractSession#complete()
         */
        @Override
        protected void complete()
        {
            super.complete();

            //lock so that no other thread that might be calling access can proceed until this complete is done
            _lock.lock();

            try
            {
                //if this is the last request thread to be in the session
                if (_activeThreads.decrementAndGet() == 0)
                {
                    try
                    {
                        //an invalid session will already have been removed from the
                        //local session map and deleted from the cluster. If its valid save
                        //it to the cluster.
                        //TODO consider doing only periodic saves if only the last access
                        //time to the session changes
                        if (isValid())
                        {
                            //if session still valid && its dirty or stale or never been synced, write it to the cluster
                            //otherwise, we just keep the updated last access time in memory
                            if (_dirty || getLastSyncTime() == 0 || isStale(System.currentTimeMillis()))
                            {
                                willPassivate();
                                save(this);
                                didActivate();
                            }
                        }
                    }
                    catch (Exception e)
                    {
                        LOG.warn("Problem saving session({})",getId(), e);
                    } 
                    finally
                    {
                        _dirty = false;
                    }
                }
            }
            finally
            {
                _lock.unlock();
            }
        }
        
        /** Test if the session is stale
         * @param atTime time when stale
         * @return true if stale
         */
        protected boolean isStale (long atTime)
        {
            return (getStaleIntervalSec() > 0) && (atTime - getLastSyncTime() >= (getStaleIntervalSec()*1000L));
        }
        
        
        /** Test if the session is dirty
         * @return true if dirty
         */
        protected boolean isDirty ()
        {
            return _dirty;
        }

        /** 
         * Expire the session.
         * 
         * @see org.eclipse.jetty.server.session.AbstractSession#timeout()
         */
        @Override
        protected void timeout()
        {
            super.timeout();
        }
        
      
        /**
         * Reload the session from the cluster. If the node that
         * last managed the session from the cluster is ourself,
         * then the session does not need refreshing.
         * NOTE: this method MUST be called with sufficient locks
         * in place to prevent 2 or more concurrent threads from
         * simultaneously updating the session.
         */
        private void refresh ()
        {
            //get fresh copy from the cluster
            Session fresh = load(makeKey(getClusterId(), _context));

            //if the session no longer exists, invalidate
            if (fresh == null)
            {
                invalidate();
                return;
            }

            //cluster copy assumed to be the same as we were the last
            //node to manage it
            if (fresh.getLastNode().equals(getLastNode()))
                return;

            setLastNode(getSessionIdManager().getWorkerName());
            
            //prepare for refresh
            willPassivate();

            //if fresh has no attributes, remove them
            if (fresh.getAttributes() == 0)
                this.clearAttributes();
            else
            {
                //reconcile attributes
                for (String key:fresh.getAttributeMap().keySet())
                {
                    Object freshvalue = fresh.getAttribute(key);

                    //session does not already contain this attribute, so bind it
                    if (getAttribute(key) == null)
                    { 
                        doPutOrRemove(key,freshvalue);
                        bindValue(key,freshvalue);
                    }
                    else //session already contains this attribute, update its value
                    {
                        doPutOrRemove(key,freshvalue);
                    }

                }
                // cleanup, remove values from session, that don't exist in data anymore:
                for (String key : getNames())
                {
                    if (fresh.getAttribute(key) == null)
                    {
                        Object oldvalue = getAttribute(key);
                        doPutOrRemove(key,null);
                        unbindValue(key,oldvalue);
                    }
                }
            }
            //finish refresh
            didActivate();
        }


        public void setExpiry (long expiry)
        {
            _expiryTime = expiry;
        }
        

        public long getExpiry ()
        {
            return _expiryTime;
        }
        
        public void swapId (String newId, String newNodeId)
        {
            //TODO probably synchronize rather than use the access/complete lock?
            _lock.lock();
            setClusterId(newId);
            setNodeId(newNodeId);
            _lock.unlock();
        }
        
        @Override
        public void setAttribute (String name, Object value)
        {
            Object old = changeAttribute(name, value);
            if (value == null && old == null)
                return; //if same as remove attribute but attribute was already removed, no change
            
           _dirty = true;
        }
        
        
        public String getContextPath()
        {
            return _contextPath;
        }


        public void setContextPath(String contextPath)
        {
            this._contextPath = contextPath;
        }


        public String getVHost()
        {
            return _vhost;
        }


        public void setVHost(String vhost)
        {
            this._vhost = vhost;
        }
        
        public String getLastNode()
        {
            return _lastNode;
        }


        public void setLastNode(String lastNode)
        {
            _lastNode = lastNode;
        }


        public long getLastSyncTime()
        {
            return _lastSyncTime;
        }


        public void setLastSyncTime(long lastSyncTime)
        {
            _lastSyncTime = lastSyncTime;
        }

    }


    /**
     * Start the session manager.
     *
     * @see org.eclipse.jetty.server.session.AbstractSessionManager#doStart()
     */
    @Override
    public void doStart() throws Exception
    {
        if (_sessionIdManager == null)
            throw new IllegalStateException("No session id manager defined");
        
        if (_cache == null)
            throw new IllegalStateException("No session cache defined");
        
        _sessions = new ConcurrentHashMap<String, Session>();

        //try and use a common scheduler, fallback to own
        _scheduler = getSessionHandler().getServer().getBean(Scheduler.class);
        if (_scheduler == null)
        {
            _scheduler = new ScheduledExecutorScheduler();
            _ownScheduler = true;
            _scheduler.start();
        }
        else if (!_scheduler.isStarted())
            throw new IllegalStateException("Shared scheduler not started");
 
        setScavengeInterval(getScavengeInterval());
        
        super.doStart();
    }


    /**
     * Stop the session manager.
     *
     * @see org.eclipse.jetty.server.session.AbstractSessionManager#doStop()
     */
    @Override
    public void doStop() throws Exception
    {
        super.doStop();

        if (_task!=null)
            _task.cancel();
        _task=null;
        if (_ownScheduler && _scheduler !=null)
            _scheduler.stop();
        _scheduler = null;
        
        _sessions.clear();
        _sessions = null;
    }
    
    
    /**
     * Look for sessions in local memory that have expired.
     */
    /**
     * 
     */
    public void scavenge ()
    {
        Set<String> candidateIds = new HashSet<String>();
        long now = System.currentTimeMillis();
        
        LOG.info("SessionManager for context {} scavenging at {} ", getContextPath(getContext()), now);
        for (Map.Entry<String, Session> entry:_sessions.entrySet())
        {
            long expiry = entry.getValue().getExpiry();
            if (expiry > 0 && expiry < now)
                candidateIds.add(entry.getKey());
        }

        for (String candidateId:candidateIds)
        {
            if (LOG.isDebugEnabled())
                LOG.debug("Session {} expired ", candidateId);
            
            Session candidateSession = _sessions.get(candidateId);
            if (candidateSession != null)
            {
                //double check the state of the session in the cache, as the
                //session may have migrated to another node. This leaves a window
                //where the cached session may have been changed by another node
                Session cachedSession = load(makeKey(candidateId, _context));
                if (cachedSession == null)
                {
                   if (LOG.isDebugEnabled()) LOG.debug("Locally expired session({}) does not exist in cluster ",candidateId);
                    //the session no longer exists, do a full invalidation
                    candidateSession.timeout();
                }
                else if (getSessionIdManager().getWorkerName().equals(cachedSession.getLastNode()))
                {
                    if (LOG.isDebugEnabled()) LOG.debug("Expiring session({}) local to session manager",candidateId);
                    //if I am the master of the session then it can be timed out
                    candidateSession.timeout();
                }
                else
                {
                    //some other node is the master of the session, simply remove it from my memory
                    if (LOG.isDebugEnabled()) LOG.debug("Session({}) not local to this session manager, removing from local memory", candidateId);
                    candidateSession.willPassivate();
                    _sessions.remove(candidateSession.getClusterId());
                }

            }
        }
    }
    
    
    public long getScavengeInterval ()
    {
        return _scavengeIntervalMs/1000;
    }

    
    /**
     * Set the interval between runs of the scavenger. It should not be run too
     * often.
     * 
     * 
     * @param sec scavenge interval in seconds
     */
    public void setScavengeInterval (long sec)
    {
        if (sec<=0)
            sec=60;

        long old_period=_scavengeIntervalMs;
        long period=sec*1000L;

        _scavengeIntervalMs=period;

        //add a bit of variability into the scavenge time so that not all
        //nodes with the same scavenge time sync up
        long tenPercent = _scavengeIntervalMs/10;
        if ((System.currentTimeMillis()%2) == 0)
            _scavengeIntervalMs += tenPercent;

        if (LOG.isDebugEnabled())
            LOG.debug("Scavenging every "+_scavengeIntervalMs+" ms");
        
        synchronized (this)
        {
            if (_scheduler != null && (period!=old_period || _task==null))
            {
                if (_task!=null)
                    _task.cancel();
                if (_scavenger == null)
                    _scavenger = new Scavenger();
                
                _task = _scheduler.schedule(_scavenger,_scavengeIntervalMs,TimeUnit.MILLISECONDS);
            }
        }
    }
    
    
    /**
     * Get the clustered cache instance.
     * 
     * @return the cache
     */
    public BasicCache<String, Object> getCache() 
    {
        return _cache;
    }

    
    /**
     * Set the clustered cache instance.
     * 
     * @param cache the cache
     */
    public void setCache (BasicCache<String, Object> cache) 
    {
        this._cache = cache;
    }


    public long getStaleIntervalSec()
    {
        return _staleIntervalSec;
    }


    public void setStaleIntervalSec(long staleIntervalSec)
    {
        _staleIntervalSec = staleIntervalSec;
    }


    /** 
     * Add a new session for the context related to this session manager
     * 
     * @see org.eclipse.jetty.server.session.AbstractSessionManager#addSession(org.eclipse.jetty.server.session.AbstractSession)
     */
    @Override
    protected void addSession(AbstractSession session)
    {
        if (session==null)
            return;
        
        if (LOG.isDebugEnabled()) LOG.debug("Adding session({}) to session manager for context {} on worker {}",session.getClusterId(), getContextPath(getContext()),getSessionIdManager().getWorkerName() + " with lastnode="+((Session)session).getLastNode());
        _sessions.put(session.getClusterId(), (Session)session);
        
        try
        {     
                session.willPassivate();
                save(((InfinispanSessionManager.Session)session));
                session.didActivate();
            
        }
        catch (Exception e)
        {
            LOG.warn("Unable to store new session id="+session.getId() , e);
        }
    }

    /** 
     * Ask the cluster for the session.
     * 
     * @see org.eclipse.jetty.server.session.AbstractSessionManager#getSession(java.lang.String)
     */
    @Override
    public AbstractSession getSession(String idInCluster)
    {
        Session session = null;

        //try and find the session in this node's memory
        Session memSession = (Session)_sessions.get(idInCluster);

        if (LOG.isDebugEnabled())
            LOG.debug("getSession({}) {} in session map",idInCluster,(memSession==null?"not":""));

        long now = System.currentTimeMillis();
        try
        {
            //if the session is not in this node's memory, then load it from the cluster cache
            if (memSession == null)
            {
                if (LOG.isDebugEnabled())
                    LOG.debug("getSession({}): loading session data from cluster", idInCluster);

                session = load(makeKey(idInCluster, _context));
                if (session != null)
                {
                    //We retrieved a session with the same key from the database

                    //Check that it wasn't expired
                    if (session.getExpiry() > 0 && session.getExpiry() <= now)
                    {
                        if (LOG.isDebugEnabled()) LOG.debug("getSession ({}): Session expired", idInCluster);
                        //ensure that the session id for the expired session is deleted so that a new session with the 
                        //same id cannot be created (because the idInUse() test would succeed)
                        ((InfinispanSessionIdManager)getSessionIdManager()).removeSession(session);
                        return null;  
                    }

                    //Update the last worker node to me
                    session.setLastNode(getSessionIdManager().getWorkerName());                            
                    //TODO consider saving session here if lastNode was not this node

                    //Check that another thread hasn't loaded the same session
                    Session existingSession = _sessions.putIfAbsent(idInCluster, session);
                    if (existingSession != null)
                    {
                        //use the one that the other thread inserted
                        session = existingSession;
                        LOG.debug("getSession({}): using session loaded by another request thread ", idInCluster);
                    }
                    else
                    {
                        //indicate that the session was reinflated
                        session.didActivate();
                        LOG.debug("getSession({}): loaded session from cluster", idInCluster);
                    }
                    return session;
                }
                else
                {
                    //The requested session does not exist anywhere in the cluster
                    LOG.debug("getSession({}): No session in cluster matching",idInCluster);
                    return null;
                }
            }
            else
            {
               //The session exists in this node's memory
               LOG.debug("getSession({}): returning session from local memory ", memSession.getClusterId());
                return memSession;
            }
        }
        catch (Exception e)
        {
            LOG.warn("Unable to load session="+idInCluster, e);
            return null;
        }
    }
    
    
    /** 
     * The session manager is stopping.
     * 
     * @see org.eclipse.jetty.server.session.AbstractSessionManager#shutdownSessions()
     */
    @Override
    protected void shutdownSessions() throws Exception
    {
        Set<String> keys = new HashSet<String>(_sessions.keySet());
        for (String key:keys)
        {
            Session session = _sessions.remove(key); //take the session out of the session list
            //If the session is dirty, then write it to the cluster.
            //If the session is simply stale do NOT write it to the cluster, as some other node
            //may have started managing that session - this means that the last accessed/expiry time
            //will not be updated, meaning it may look like it can expire sooner than it should.
            try
            {
                if (session.isDirty())
                {
                    if (LOG.isDebugEnabled())
                        LOG.debug("Saving dirty session {} before exiting ", session.getId());
                    save(session);
                }
            }
            catch (Exception e)
            {
                LOG.warn(e);
            }
        }
    }


    @Override
    protected AbstractSession newSession(HttpServletRequest request)
    {
        return new Session(request);
    }

    /** 
     * Remove a session from local memory, and delete it from
     * the cluster cache.
     * 
     * @see org.eclipse.jetty.server.session.AbstractSessionManager#removeSession(java.lang.String)
     */
    @Override
    protected boolean removeSession(String idInCluster)
    {
        Session session = (Session)_sessions.remove(idInCluster);
        try
        {
            if (session != null)
                delete(session);
        }
        catch (Exception e)
        {
            LOG.warn("Problem deleting session id="+idInCluster, e);
        }
        return session!=null;
    }
    
    
    @Override
    public void renewSessionId(String oldClusterId, String oldNodeId, String newClusterId, String newNodeId)
    {
        Session session = null;
        try
        {
            //take the session with that id out of our managed list
            session = (Session)_sessions.remove(oldClusterId);
            if (session != null)
            {
                //TODO consider transactionality and ramifications if the session is live on another node
                delete(session); //delete the old session from the cluster  
                session.swapId(newClusterId, newNodeId); //update the session
                _sessions.put(newClusterId, session); //put it into managed list under new key
                save(session); //put the session under the new id into the cluster
            }
        }
        catch (Exception e)
        {
            LOG.warn(e);
        }

        super.renewSessionId(oldClusterId, oldNodeId, newClusterId, newNodeId);
    }


    /**
     * Load a session from the clustered cache.
     * 
     * @param key the session key
     * @return the session
     */
    protected Session load (String key)
    {
        if (_cache == null)
            throw new IllegalStateException("No cache");
        
        if (LOG.isDebugEnabled()) LOG.debug("Loading session {} from cluster", key);

        SerializableSessionData storableSession = (SerializableSessionData)_cache.get(key);
        if (storableSession == null)
        {
            if (LOG.isDebugEnabled()) LOG.debug("No session {} in cluster ",key);
            return null;
        }
        else
        {
            Session session = new Session (storableSession);
            session.setLastSyncTime(System.currentTimeMillis());
            return session;
        }
    }
    
    
    /**
     * Save or update the session to the cluster cache
     * 
     * @param session the session
     * @throws Exception if unable to save
     */
    protected void save (InfinispanSessionManager.Session session)
    throws Exception
    {
        if (_cache == null)
            throw new IllegalStateException("No cache");
        
        if (LOG.isDebugEnabled()) LOG.debug("Writing session {} to cluster", session.getId());
    
        SerializableSessionData storableSession = new SerializableSessionData(session);

        //Put an idle timeout on the cache entry if the session is not immortal - 
        //if no requests arrive at any node before this timeout occurs, or no node 
        //scavenges the session before this timeout occurs, the session will be removed.
        //NOTE: that no session listeners can be called for this.
        InfinispanSessionIdManager sessionIdManager = (InfinispanSessionIdManager)getSessionIdManager();
        if (storableSession.maxInactive > 0)
            _cache.put(makeKey(session, _context), storableSession, -1, TimeUnit.SECONDS, storableSession.maxInactive*sessionIdManager.getIdleExpiryMultiple(), TimeUnit.SECONDS);
        else
            _cache.put(makeKey(session, _context), storableSession);
        
        //tickle the session id manager to keep the sessionid entry for this session up-to-date
        sessionIdManager.touch(session.getClusterId());
        
        session.setLastSyncTime(System.currentTimeMillis());
    }
    
    
    /**
     * Remove the session from the cluster cache.
     * 
     * @param session the session
     */
    protected void delete (InfinispanSessionManager.Session session)
    {  
        if (_cache == null)
            throw new IllegalStateException("No cache");
        if (LOG.isDebugEnabled()) LOG.debug("Removing session {} from cluster", session.getId());
        _cache.remove(makeKey(session, _context));
    }

    
    /**
     * Invalidate a session for this context with the given id
     * 
     * @param idInCluster session id in cluster
     */
    public void invalidateSession (String idInCluster)
    {
        Session session = (Session)_sessions.get(idInCluster);

        if (session != null)
        {
            session.invalidate();
        }
    }

    
    /**
     * Make a unique key for this session.
     * As the same session id can be used across multiple contexts, to
     * make it unique, the key must be composed of:
     * <ol>
     * <li>the id</li>
     * <li>the context path</li>
     * <li>the virtual hosts</li>
     * </ol>
     * 
     *TODO consider the difference between getClusterId and getId
     * @param session
     * @return
     */
    private String makeKey (Session session, Context context)
    {
       return makeKey(session.getId(), context);
    }
    
    /**
     * Make a unique key for this session.
     * As the same session id can be used across multiple contexts, to
     * make it unique, the key must be composed of:
     * <ol>
     * <li>the id</li>
     * <li>the context path</li>
     * <li>the virtual hosts</li>
     * </ol>
     * 
     *TODO consider the difference between getClusterId and getId
     * @param session
     * @return
     */
    private String makeKey (String id, Context context)
    {
        String key = getContextPath(context);
        key = key + "_" + getVirtualHost(context);
        key = key+"_"+id;
        return key;
    }
    
    /**
     * Turn the context path into an acceptable string
     * 
     * @param context
     * @return
     */
    private static String getContextPath (ContextHandler.Context context)
    {
        return canonicalize (context.getContextPath());
    }

    /**
     * Get the first virtual host for the context.
     *
     * Used to help identify the exact session/contextPath.
     *
     * @return 0.0.0.0 if no virtual host is defined
     */
    private static String getVirtualHost (ContextHandler.Context context)
    {
        String vhost = "0.0.0.0";

        if (context==null)
            return vhost;

        String [] vhosts = context.getContextHandler().getVirtualHosts();
        if (vhosts==null || vhosts.length==0 || vhosts[0]==null)
            return vhost;

        return vhosts[0];
    }

    /**
     * Make an acceptable name from a context path.
     *
     * @param path
     * @return
     */
    private static String canonicalize (String path)
    {
        if (path==null)
            return "";

        return path.replace('/', '_').replace('.','_').replace('\\','_');
    }

}