public HASessionStateImpl(String sessionStateName,
    HAPartition partition,
    long beanCleaningDelay) 
{
   
      if (partition == null)
      {
         throw new IllegalArgumentException("HAPartition must not be null when constructing HASessionImpl");
      }
      this.hapGeneral = partition;
      if (sessionStateName == null)
      {
         this._sessionStateName = org.jboss.metadata.ClusterConfigMetaData.DEFAULT_SESSION_STATE_NAME;
      }
      else
      {
         this._sessionStateName = sessionStateName;
      }
      this.sessionStateIdentifier = "SessionState-'" + this._sessionStateName + "'";
      if (beanCleaningDelay  > 0)
      {
         this.beanCleaningDelay = beanCleaningDelay;
      }
      else
      {
         this.beanCleaningDelay = MAX_DELAY_BEFORE_CLEANING_UNRECLAIMED_STATE;
      }
}

 public PackagedSessionImpl _createSession(String appName,
    Object keyId) 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      PackagedSessionImpl result = new PackagedSessionImpl((Serializable) keyId, null, this.myNodeName);
      app.put(keyId, result);
      return result;
}

 public  void _removeSession(String appName,
    Object keyId) 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      PackagedSession ps = app.remove(keyId);
      if (ps != null && ps.getOwner().equals(this.myNodeName))
      {
         this.ownedObjectExternallyModified(appName, keyId, ps, ps);
      }
      this.removeLock(appName, keyId);
}

 public Boolean _setOwnership(String appName,
    Object keyId,
    String newOwner,
    Long remoteVersion) 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      PackagedSession ps = app.get(keyId);
      Boolean answer = Boolean.TRUE;
      Mutex mtx = this.getLock(appName, keyId);
      try
      {
         if (!mtx.attempt(0))
         {
            return Boolean.FALSE;
         }
      }
      catch (InterruptedException ie)
      {
         this.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);
            this.ownedObjectExternallyModified(appName, keyId, ps, ps);
            answer = Boolean.TRUE;
         }
      }
      finally
      {
         mtx.release();
      }
      return answer;
}

 public  void _setState(String appName,
    PackagedSession session) 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      PackagedSession ps = app.get(session.getKey());
      if (ps == null)
      {
         ps = session;
         synchronized (app)
         {
            app.put(ps.getKey(), ps);
         }
      }
      else
      {
         Mutex mtx = this.getLock(appName, session.getKey());
         try
         {
            mtx.acquire();
         }
         catch (InterruptedException ie)
         {
            this.log.info(ie);
            return;
         }
         try
         {
            if (ps.getOwner().equals(this.myNodeName))
            {
               // a modification has occured externally while we were the owner
               //
               this.ownedObjectExternallyModified(appName, session.getKey(), ps, session);
            }
            ps.update(session);
         }
         finally
         {
            mtx.release();
         }
      }
}

 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
      //
      org.jboss.util.naming.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)
         {
            this.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 createSession(String appName,
    Object keyId) 
{
      this._createSession(appName, keyId);
}

 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();
}

 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 HAPartition getCurrentHAPartition() 
{
      return this.hapGeneral;
}

 public Serializable getCurrentState() 
{
      this.log.debug ("Building and returning state of HASessionState");
      if (this.appSessions == null)
      {
         this.appSessions = new Hashtable< String, Hashtable< Object, PackagedSession > >();
      }
      Serializable result = null;
      synchronized (this.lockAppSession)
      {
         this.purgeState();
         try
         {
            result = this.deflate(this.appSessions);
         }
         catch (Exception e)
         {
            this.log.error("operation failed", e);
         }
      }
      return result;
}

 protected Hashtable getHashtableForApp(String appName) 
{
   
      if (this.appSessions == null)
      {
         this.appSessions = new Hashtable< String, Hashtable< Object, PackagedSession > >(); // should never happen though...
      }
      Hashtable< Object, PackagedSession > result = null;
      synchronized (this.lockAppSession)
      {
         result = this.appSessions.get(appName);
         if (result == null)
         {
            result = new Hashtable< Object, PackagedSession >();
            this.appSessions.put(appName, result);
         }
      }
      return result;
}

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

 public String getNodeName() 
{
      return this.myNodeName;
}

 public PackagedSession getState(String appName,
    Object keyId) 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      return app.get(keyId);
}

 public PackagedSession getStateWithOwnership(String appName,
    Object keyId) throws RemoteException 
{
      return this.localTakeOwnership(appName, keyId);
}

 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());
      }
}

 public  void init() throws Exception 
{
      this.log = Logger.getLogger(HASessionStateImpl.class.getName() + "." + this._sessionStateName);
      this.hapGeneral.registerRPCHandler(this.sessionStateIdentifier, this);
      this.hapGeneral.subscribeToStateTransferEvents(HA_SESSION_STATE_STATE_TRANSFER, this);
}

 public PackagedSession localTakeOwnership(String appName,
    Object keyId) throws RemoteException 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      PackagedSession ps = 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 = this.getLock (appName, keyId);
      try
      {
         if (!mtx.attempt(0))
         {
            throw new java.rmi.RemoteException("Concurent calls on session object.");
         }
      }
      catch (InterruptedException ie)
      {
         this.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,
                                                             SET_OWNERSHIP_TYPES,
                                                             true);
            }
            catch (Exception e)
            {
               this.log.error("operation failed", e);
            }
            if ((answers != null) && answers.contains(Boolean.FALSE))
            {
               throw new java.rmi.RemoteException("Concurent calls on session object.");
            }
            ps.setOwner(this.myNodeName);
            return ps;
         }
         return ps;
      }
      finally
      {
         mtx.release();
      }
}

 protected boolean lockExists(String appName,
    Object key) 
{
      synchronized (this.locks)
      {
         Map< Object, Mutex > ls = this.locks.get(appName);
         if (ls == null)
         {
            return false;
         }
         return (ls.get(key) != null);
      }
}

 public  void ownedObjectExternallyModified(String appName,
    Object key,
    PackagedSession oldSession,
    PackagedSession newSession) 
{
      Vector< HASessionStateListener > members = this.listeners.get(appName);
      if (members != null)
      {
         for (int i = 0; i <  members.size(); i++)
         {
            try
            {
               members.elementAt(i).sessionExternallyModified(newSession);
            }
            catch (Throwable t)
            {
               this.log.debug(t);
            }
         }
      }
}

 public  void purgeState() 
{
      synchronized (this.lockAppSession)
      {
         for (Enumeration< String > keyEnum = this.appSessions.keys(); keyEnum.hasMoreElements();)
         {
            // trip in apps..
            //
            String key = keyEnum.nextElement();
            Hashtable< Object, PackagedSession > value = this.appSessions.get(key);
            long currentTime = System.currentTimeMillis();
            for (Iterator< PackagedSession > iterSessions = value.values().iterator(); iterSessions.hasNext();)
            {
               PackagedSession ps = iterSessions.next();
               if ((currentTime - ps.unmodifiedExistenceInVM())  > this.beanCleaningDelay)
               {
                  iterSessions.remove();
               }
            }
         }
      }
}

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

 public  void removeSession(String appName,
    Object keyId) 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      if (app != null)
      {
         PackagedSession ps = app.remove(keyId);
         if (ps != null)
         {
            this.removeLock(appName, keyId);
            Object[] args = { appName, keyId };
            try
            {
               this.hapGeneral.callMethodOnCluster (this.sessionStateIdentifier,
                                                    "_removeSession",
                                                    args,
                                                    REMOVE_SESSION_TYPES,
                                                    true);
            }
            catch (Exception e)
            {
               this.log.error("operation failed", e);
            }
         }
      }
}

 public  void setCurrentState(Serializable newState) 
{
      this.log.debug("Receiving state of HASessionState");
      if (this.appSessions == null)
      {
         this.appSessions = new Hashtable< String, Hashtable< Object, PackagedSession > >();
      }
      synchronized (this.lockAppSession)
      {
         try
         {
            this.appSessions.clear(); // hope to facilitate the job of the GC
            this.appSessions = (Hashtable< String, Hashtable< Object, PackagedSession > >) this.inflate((byte[]) newState);
         }
         catch (Exception e)
         {
            this.log.error("operation failed", e);
         }
      }
}

 public  void setState(String appName,
    Object keyId,
    byte[] state) throws RemoteException 
{
      Hashtable< Object, PackagedSession > app = this.getHashtableForApp(appName);
      PackagedSession ps = app.get(keyId);
      if (ps == null)
      {
         ps = this._createSession(appName, keyId);
      }
      boolean isStateIdentical = false;
      Mutex mtx = this.getLock(appName, keyId);
      try
      {
         if (!mtx.attempt(0))
         {
            throw new java.rmi.RemoteException ("Concurent calls on session object.");
         }
      }
      catch (InterruptedException ie)
      {
         this.log.info(ie);
         return;
      }
      try
      {
         isStateIdentical = ps.setState(state);
         if (!isStateIdentical)
         {
            Object[] args = { appName, ps };
            try
            {
               this.hapGeneral.callMethodOnCluster(this.sessionStateIdentifier,
                                                   "_setState",
                                                   args,
                                                   SET_STATE_TYPES,
                                                   true);
            }
            catch (Exception e)
            {
               this.log.error("operation failed", e);
            }
         }
      }
      finally
      {
         mtx.release();
      }
}

 public  void start() throws Exception 
{
      this.myNodeName = this.hapGeneral.getNodeName();
      this.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);
}

 public  void stop() 
{
      this.purgeState();
      // Unbind so we can rebind if restarted
      try
      {
         Context ctx = new InitialContext();
         ctx.unbind(this._sessionStateName);
         org.jboss.util.naming.NonSerializableFactory.unbind(this._sessionStateName);
      }
      catch (NamingException e)
      {
         // Ignore
      }
}

 public synchronized  void subscribe(String appName,
    HASessionStateListener listener) 
{
   
      Vector< HASessionStateListener > members = this.listeners.get(appName);
      if (members == null)
      {
         members = new Vector< HASessionStateListener >();
         this.listeners.put(appName, members);
      }
      if (!members.contains(listener))
      {
         members.add(listener);
      }
}

 public  void takeOwnership(String appName,
    Object keyId) throws RemoteException 
{
      this.localTakeOwnership(appName, keyId);
}

 public synchronized  void unsubscribe(String appName,
    HASessionStateListener listener) 
{
      Vector< HASessionStateListener > members = this.listeners.get(appName);
      if ((members != null) && members.contains(listener))
      {
         members.remove(listener);
      }
}