/*
 * JBoss, Home of Professional Open Source.
 * Copyright 2006, Red Hat Middleware LLC, and individual contributors
 * as indicated by the @author tags. See the copyright.txt file in the
 * distribution for a full listing of individual contributors.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this software; if not, write to the Free
 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
 */
package org.jboss.ha.hasessionstate.server;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Vector;
import java.util.zip.Deflater;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.Name;
import javax.naming.NameNotFoundException;
import javax.naming.Reference;
import javax.naming.StringRefAddr;

import org.jboss.ha.framework.interfaces.HAPartition;
import org.jboss.ha.hasessionstate.interfaces.PackagedSession;
import org.jboss.logging.Logger;
import org.jboss.naming.NonSerializableFactory;
import org.jboss.system.server.ServerConfigUtil;

import EDU.oswego.cs.dl.util.concurrent.Mutex;

/**
 *   Default implementation of HASessionState
 *
 *   @see org.jboss.ha.hasessionstate.interfaces.HASessionState
 *   @author sacha.labourey@cogito-info.ch
 *   @author <a href="bill@burkecentral.com">Bill Burke</a>
 *   @version $Revision: 57188 $
 *
 * <p><b>Revisions:</b><br>
 * <p><b>2002/01/09: billb</b>
 * <ol>
 *   <li>ripped out sub partitioning stuff.  It really belongs as a subclass of HAPartition
 * </ol>
 * 
 */

public class HASessionStateImpl
   implements org.jboss.ha.hasessionstate.interfaces.HASessionState,
	      HAPartition.HAPartitionStateTransfer
{
   
   protected String _sessionStateName;
   protected Logger log;
   protected HAPartition hapGeneral;
   protected String sessionStateIdentifier;
   protected String myNodeName;
   
   protected long beanCleaningDelay;
   protected String haPartitionName;
   protected String haPartitionJndiName;
   
   protected final String DEFAULT_PARTITION_JNDI_NAME = ServerConfigUtil.getDefaultPartitionName();
   protected final String JNDI_FOLDER_NAME_FOR_HASESSIONSTATE = org.jboss.metadata.ClusterConfigMetaData.JNDI_PREFIX_FOR_SESSION_STATE;
   protected final String JNDI_FOLDER_NAME_FOR_HAPARTITION = "/HAPartition/";
   protected final long MAX_DELAY_BEFORE_CLEANING_UNRECLAIMED_STATE = 30L * 60L * 1000L; // 30 minutes... should be set externally or use cache settings
   protected static final String HA_SESSION_STATE_STATE_TRANSFER = "HASessionStateTransfer";
   
   protected HashMap locks = new HashMap ();
      
   public HASessionStateImpl ()
   {}
   
   public HASessionStateImpl (String sessionStateName,
                              HAPartition partition,
                              long beanCleaningDelay)
   {
      this(sessionStateName, partition.getPartitionName(), beanCleaningDelay);
      this.hapGeneral = partition;
   }
   
   public HASessionStateImpl (String sessionStateName,
                              String mainHAPartitionName,
                              long beanCleaningDelay)
   {
      if (sessionStateName == null)
         this._sessionStateName = org.jboss.metadata.ClusterConfigMetaData.DEFAULT_SESSION_STATE_NAME;
      else
         this._sessionStateName = sessionStateName;
      
      this.sessionStateIdentifier = "SessionState-'" + this._sessionStateName + "'";
      
      if (mainHAPartitionName == null)
         haPartitionName = DEFAULT_PARTITION_JNDI_NAME;
      else
         haPartitionName = mainHAPartitionName;
      
      haPartitionJndiName = JNDI_FOLDER_NAME_FOR_HAPARTITION + haPartitionName;
      
      if (beanCleaningDelay > 0)
         this.beanCleaningDelay = beanCleaningDelay;
      else
         this.beanCleaningDelay = MAX_DELAY_BEFORE_CLEANING_UNRECLAIMED_STATE;
      
   }
   
   public void init () throws Exception
   {
      this.log = Logger.getLogger(HASessionStateImpl.class.getName() + "." + this._sessionStateName);

      // BES 20060416 -- if people used an old config, we may not
      // have been passed the partition, so have to find in JNDI
      // JNDI work s/b done in start(), but we have no choice, as
      // we must register for state transfer in init
      if (this.hapGeneral == null)
      {
         Context ctx = new InitialContext ();      
         this.hapGeneral = (HAPartition)ctx.lookup (haPartitionJndiName);
      }
      
      if (hapGeneral == null)
         log.error ("Unable to get default HAPartition under name '" + haPartitionJndiName + "'.");
      
      this.hapGeneral.registerRPCHandler (this.sessionStateIdentifier, this);
      this.hapGeneral.subscribeToStateTransferEvents (HA_SESSION_STATE_STATE_TRANSFER, this);
   }
   
   public void start () throws Exception
   {
      this.myNodeName = this.hapGeneral.getNodeName ();
      log.debug ("HASessionState node name : " + this.myNodeName );
      
      // BES 4/7/06 clean up lifecycle; move this to start, as it can't be 
      // called until startService due to JNDI dependency
      Context ctx = new InitialContext ();
      this.bind (this._sessionStateName, this, HASessionStateImpl.class, ctx);
   }
   
   protected void bind (String jndiName, Object who, Class classType, Context ctx) throws Exception
   {
      // Ah ! This service isn't serializable, so we use a helper class
      //
      NonSerializableFactory.bind (jndiName, who);
      Name n = ctx.getNameParser ("").parse (jndiName);
      while (n.size () > 1)
      {
         String ctxName = n.get (0);
         try
         {
            ctx = (Context)ctx.lookup (ctxName);
         }
         catch (NameNotFoundException e)
         {
            log.debug ("creating Subcontext" + ctxName);
            ctx = ctx.createSubcontext (ctxName);
         }
         n = n.getSuffix (1);
      }
      
      // The helper class NonSerializableFactory uses address type nns, we go on to
      // use the helper class to bind the service object in JNDI
      //
      StringRefAddr addr = new StringRefAddr ("nns", jndiName);
      Reference ref = new Reference ( classType.getName (), addr, NonSerializableFactory.class.getName (), null);
      ctx.bind (n.get (0), ref);
   }
   
   public void stop () throws Exception
   {      
      purgeState();
       
      // Unbind so we can rebind if restarted
      try
      {
         Context ctx = new InitialContext ();         
         ctx.unbind (this._sessionStateName);
         NonSerializableFactory.unbind (this._sessionStateName);
      }
      catch (Exception ignored)
      {}
   }
   
   public void destroy() throws Exception
   {       
      // Remove ref to ourself from HAPartition
      this.hapGeneral.unregisterRPCHandler(this.sessionStateIdentifier, this);
      this.hapGeneral.unsubscribeFromStateTransferEvents(HA_SESSION_STATE_STATE_TRANSFER, this);  
   }
   
   public String getNodeName ()
   {
      return this.myNodeName ;
   }
   
   // Used for Session state transfer
   //
   public Serializable getCurrentState ()
   {
      log.debug ("Building and returning state of HASessionState");
      
      if (this.appSessions == null)
         this.appSessions = new Hashtable ();
      
      Serializable result = null;
      
      synchronized (this.lockAppSession)
      {
         this.purgeState ();
         
         try
         {
            result = deflate (this.appSessions);
         }
         catch (Exception e)
         {
            log.error("operation failed", e);
         }
      }
      return result;
   }
   
   public void setCurrentState (Serializable newState)
   {
      log.debug ("Receiving state of HASessionState");
      
      if (this.appSessions == null)
         this.appSessions = new Hashtable ();
      
      synchronized (this.lockAppSession)
      {
         try
         {
            this.appSessions.clear (); // hope to facilitate the job of the GC
            this.appSessions = (Hashtable)inflate ((byte[])newState);
         }
         catch (Exception e)
         {
            log.error("operation failed", e);
         }         
      }
   }
   
   public void purgeState ()
   {
      synchronized (this.lockAppSession)
      {
         for (Enumeration keyEnum = this.appSessions.keys (); keyEnum.hasMoreElements ();)
         {
            // trip in apps..
            //
            Object key = keyEnum.nextElement ();
            Hashtable value = (Hashtable)this.appSessions.get (key);
            long currentTime = System.currentTimeMillis ();
            
            for (Iterator iterSessions = value.values ().iterator (); iterSessions.hasNext ();)
            {
               PackagedSession ps = (PackagedSession)iterSessions.next ();
               if ( (currentTime - ps.unmodifiedExistenceInVM ()) > beanCleaningDelay )
                  iterSessions.remove ();
            }
         }
      }
      
   }
   
   protected byte[] deflate (Object object) throws IOException
   {
      ByteArrayOutputStream baos = new ByteArrayOutputStream ();
      Deflater def = new Deflater (java.util.zip.Deflater.BEST_COMPRESSION);
      DeflaterOutputStream dos = new DeflaterOutputStream (baos, def);
      
      ObjectOutputStream out = new ObjectOutputStream (dos);
      out.writeObject (object);
      out.close ();
      dos.finish ();
      dos.close ();
      
      return baos.toByteArray ();
   }
   
   protected Object inflate (byte[] compressedContent) throws IOException
   {
      if (compressedContent==null)
         return null;
      
      try
      {
         ObjectInputStream in = new ObjectInputStream (new InflaterInputStream (new ByteArrayInputStream (compressedContent)));
         
         Object object = in.readObject ();
         in.close ();
         return object;
      }
      catch (Exception e)
      {
         throw new IOException (e.toString ());
      }
   }
   
   protected Hashtable appSessions = new Hashtable ();
   protected Object lockAppSession = new Object ();
   
   protected Hashtable getHashtableForApp (String appName)
   {
      if (this.appSessions == null)
         this.appSessions = new Hashtable (); // should never happen though...
      
      Hashtable result = null;
      
      synchronized (this.lockAppSession)
      {
         result = (Hashtable)this.appSessions.get (appName);
         if (result == null)
         {
            result = new Hashtable ();
            this.appSessions.put (appName, result);
         }
      }
      return result;
   }
   
   public void createSession (String appName, Object keyId)
   {
      this._createSession (appName, keyId);
   }
   
   public PackagedSessionImpl _createSession (String appName, Object keyId)
   {
      Hashtable app = this.getHashtableForApp (appName);
      PackagedSessionImpl result = new PackagedSessionImpl ((Serializable)keyId, null, this.myNodeName);
      app.put (keyId, result);
      return result;
   }
   
   public void setState (String appName, Object keyId, byte[] state)
      throws java.rmi.RemoteException
   {
      Hashtable app = this.getHashtableForApp (appName);
      PackagedSession ps = (PackagedSession)app.get (keyId);
      
      if (ps == null)
      {
         ps = _createSession (appName, keyId);
      }
            
      boolean isStateIdentical = false;
      
      Mutex mtx = getLock (appName, keyId);
      try { 
         if (!mtx.attempt (0))
            throw new java.rmi.RemoteException ("Concurent calls on session object.");
      } 
      catch (InterruptedException ie) { log.info (ie); return; }
      
      try
      {
         isStateIdentical = ps.setState(state);
         if (!isStateIdentical)
         {
            Object[] args =
               {appName, ps};
            try
            {
               this.hapGeneral.callMethodOnCluster (this.sessionStateIdentifier,
                                                    "_setState",
                                                    args,
                                                    new Class[]{String.class, PackagedSession.class}, true);
            }
            catch (Exception e)
            {
               log.error("operation failed", e);
            }
         }
      }
      finally
      {
         mtx.release ();
      }
   }
   
   /*
   public void _setStates (String appName, Hashtable packagedSessions)
   {
      synchronized (this.lockAppSession)
      {
         Hashtable app = this.getHashtableForApp (appName);
         
         if (app == null)
         {
            app = new Hashtable (packagedSessions.size ());
            this.appSessions.put (appName, app);
         }
         app.putAll (packagedSessions);
      }
   }*/
   
   public void _setState (String appName, PackagedSession session)
   {
      Hashtable app = this.getHashtableForApp (appName);
      PackagedSession ps = (PackagedSession)app.get (session.getKey ());
      
      if (ps == null)
      {
         ps = session;
         synchronized (app)
         {
            app.put (ps.getKey (), ps);
         }
      }
      else
      {
         Mutex mtx = getLock (appName, session.getKey ());
         try { mtx.acquire (); } catch (InterruptedException ie) { log.info (ie); return; }
         
         try
         {
            if (ps.getOwner ().equals (this.myNodeName))
            {
               // a modification has occured externally while we were the owner
               //
               ownedObjectExternallyModified (appName, session.getKey (), ps, session);
            }
            ps.update (session);
         }
         finally
         {
            mtx.release ();
         }
      }
      
   }
   
   public PackagedSession getState (String appName, Object keyId)
   {
      Hashtable app = this.getHashtableForApp (appName);
      return (PackagedSession)app.get (keyId);
   }
   
   public PackagedSession getStateWithOwnership (String appName, Object keyId) throws java.rmi.RemoteException
   {
      return this.localTakeOwnership (appName, keyId);
   }
   
   public PackagedSession localTakeOwnership (String appName, Object keyId) throws java.rmi.RemoteException
   {
      Hashtable app = this.getHashtableForApp (appName);
      PackagedSession ps = (PackagedSession)app.get (keyId);
      
      // if the session is not yet available, we simply return null. The persistence manager
      // will have to take an action accordingly
      //
      if (ps == null)
         return null;
      
      Mutex mtx = getLock (appName, keyId);
      
      try { 
         if (!mtx.attempt (0))
            throw new java.rmi.RemoteException ("Concurent calls on session object.");
      } 
      catch (InterruptedException ie) { log.info (ie); return null; }
      
      try
      {
         if (!ps.getOwner ().equals (this.myNodeName))
         {
            Object[] args =
            {appName, keyId, this.myNodeName, new Long (ps.getVersion ())};
            ArrayList answers = null;
            try
            {
               answers = this.hapGeneral.callMethodOnCluster (this.sessionStateIdentifier,
                                                              "_setOwnership",
                                                              args,
                                                              new Class[]{String.class, Object.class,
                                                                          String.class, Long.class},
                                                              true);
            }
            catch (Exception e)
            {
               log.error("operation failed", e);
            }
            
            if (answers != null && answers.contains (Boolean.FALSE))
               throw new java.rmi.RemoteException ("Concurent calls on session object.");
            else
            {
               ps.setOwner (this.myNodeName);
               return ps;
            }
         }
         else
            return ps;
      }
      finally
      {
         mtx.release ();
      }
   }
   
   public Boolean _setOwnership (String appName, Object keyId, String newOwner, Long remoteVersion)
   {
      Hashtable app = this.getHashtableForApp (appName);
      PackagedSession ps = (PackagedSession)app.get (keyId);
      Boolean answer = Boolean.TRUE;
      Mutex mtx = getLock (appName, keyId);
      
      try { 
         if (!mtx.attempt (0))
            return Boolean.FALSE;
      } 
      catch (InterruptedException ie) { log.info (ie); return Boolean.FALSE; }

      try
      {
         if (!ps.getOwner ().equals (this.myNodeName))
         {
            // this is not our business... we don't care
            // we do not update the owner of ps as another host may refuse the _setOwnership call
            // anyway, the update will be sent to us later if state is modified
            //
            //ps.setOwner (newOwner);
            answer = Boolean.TRUE;
         }
         else if (ps.getVersion () > remoteVersion.longValue ())
         {
            // we are concerned and our version is more recent than the one of the remote host!
            // it means that we have concurrent calls on the same state that has not yet been updated
            // this means we will need to raise a java.rmi.RemoteException
            //
            answer = Boolean.FALSE;
         }
         else
         {
            // the remote host has the same version as us (or more recent? possible?)
            // we need to update the ownership. We can do this because we know that no other
            // node can refuse the _setOwnership call
            ps.setOwner (newOwner);
            ownedObjectExternallyModified (appName, keyId, ps, ps);
            answer = Boolean.TRUE;
         }
      }
      finally
      {
         mtx.release ();
      }
      return answer;
   }
   
   public void takeOwnership (String appName, Object keyId) throws java.rmi.RemoteException
   {
      this.localTakeOwnership (appName, keyId);
   }
   
   public void removeSession (String appName, Object keyId)
   {
      Hashtable app = this.getHashtableForApp (appName);
      if (app != null)
      {
         PackagedSession ps = (PackagedSession)app.remove (keyId);
         if (ps != null)
         {
            removeLock (appName, keyId);
            Object[] args =
               { appName, keyId };
            try
            {
               this.hapGeneral.callMethodOnCluster (this.sessionStateIdentifier,
                                                    "_removeSession",
                                                    args,
                                                    new Class[]{String.class, Object.class},
                                                    true);
            }
            catch (Exception e)
            { log.error("operation failed", e); }
         }
      }
   }
   
   public void _removeSession (String appName, Object keyId)
   {
      Hashtable app = this.getHashtableForApp (appName);
      PackagedSession ps = null;
      ps = (PackagedSession)app.remove (keyId);
      if (ps != null && ps.getOwner ().equals (this.myNodeName))
         ownedObjectExternallyModified (appName, keyId, ps, ps);
      
      removeLock (appName, keyId);
   }
   
   protected Hashtable listeners = new Hashtable ();
   
   public synchronized void subscribe (String appName, HASessionStateListener listener)
   {
      Vector members = (Vector)listeners.get (appName);
      if (members == null)
      {
         members = new Vector ();
         listeners.put (appName, members);
      }
      if (!members.contains (listener))
      {
         members.add (listener);
      }

   }
   
   public synchronized void unsubscribe (String appName, HASessionStateListener listener)
   {
      Vector members = (Vector)listeners.get (appName);
      if ((members != null) && members.contains (listener))
         members.remove (listener);
   }
   
   public void ownedObjectExternallyModified (String appName, Object key, PackagedSession oldSession, PackagedSession newSession)
   {
      Vector members = (Vector)listeners.get (appName);
      if (members != null)
         for (int i=0; i<members.size (); i++)
         try
         {
            ((HASessionStateListener)members.elementAt (i)).sessionExternallyModified (newSession);
         }
         catch (Throwable t)
         {
            log.debug (t);
         }
   }
   
   public HAPartition getCurrentHAPartition ()
   {
      return this.hapGeneral;
   }
   
   
   protected boolean lockExists (String appName, Object key)
   {
      synchronized (this.locks)
      {
         HashMap ls = (HashMap)this.locks.get (appName);
         if (ls == null)
            return false;
         
         return (ls.get(key)!=null);
      }
   }

   protected Mutex getLock (String appName, Object key)
   {
      synchronized (this.locks)
      {
         HashMap ls = (HashMap)this.locks.get (appName);
         if (ls == null)
         {
            ls = new HashMap ();
            this.locks.put (appName, ls);
         }
          
         Mutex mutex = (Mutex)ls.get(key);
         if (mutex == null)
         {
            mutex = new Mutex ();
            ls.put (key, mutex);
         }
         
         return mutex;         
      }
   }

   protected void removeLock (String appName, Object key)
   {
      synchronized (this.locks)
      {
         HashMap ls = (HashMap)this.locks.get (appName);
         if (ls == null)
            return;
         ls.remove (key);
      }
   }
   
}