/*
* JBoss, Home of Professional Open Source
* Copyright 2005, JBoss Inc., and individual contributors as indicated
* by the @authors tag. See the copyright.txt 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.framework.server;


import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import javax.management.MBeanServer;
import javax.management.ObjectName;

import org.jboss.ha.framework.interfaces.HAPartition;
import org.jboss.ha.framework.interfaces.HAPartition.HAPartitionStateTransfer;
import org.jboss.logging.Logger;
import org.jboss.system.Registry;

/**
 *   This class manages distributed state across the cluster.
 *
 * @author  <a href="mailto:sacha.labourey@cogito-info.ch">Sacha Labourey</a>.
 * @author  <a href="mailto:bill@burkecentral.com">Bill Burke</a>.
 * @author  Scott.Stark@jboss.org
 * @version $Revision$
 */
public class DistributedStateImpl
   implements DistributedStateImplMBean, HAPartitionStateTransfer
{
   // Constants -----------------------------------------------------

   protected final static String SERVICE_NAME = "DistributedState";

   protected final static Class[] set_types=new Class[]{String.class, Serializable.class, Serializable.class};
   protected final static Class[] remove_types=new Class[]{String.class, Serializable.class};

   // Attributes ----------------------------------------------------

   /**
    * HashMap<String, HashMap>. Keys= category, value = HashMap<Object, Object>
    */
   protected HashMap categories = new HashMap ();


   protected HashMap keyListeners = new HashMap ();
   protected HAPartition partition;
   protected Logger log;
   protected MBeanServer mbeanServer = null;
   protected String name = null;

   // Static --------------------------------------------------------c

   // Constructors --------------------------------------------------

   public DistributedStateImpl () {} // for JMX checks

   public DistributedStateImpl (HAPartition partition, MBeanServer server)
   {
      this.partition = partition;
      this.mbeanServer = server;
      this.log = Logger.getLogger (this.getClass ());
   }

   // Public --------------------------------------------------------

   public void init () throws Exception
   {
      // When we subscribe to state transfer events, GetState will be called to initialize
      // this service.
      partition.subscribeToStateTransferEvents (SERVICE_NAME, this);
      partition.registerRPCHandler (SERVICE_NAME, this);

      // subscribed this "sub-service" of HAPartition with JMX
      // TODO: In the future (when state transfer issues will be completed),
      // we will need to redesign the way HAPartitions and its sub-protocols are
      // registered with JMX. They will most probably be independant JMX services.
      //
      this.name = "jboss:service=" + SERVICE_NAME +
                    ",partitionName=" + this.partition.getPartitionName();
      ObjectName jmxName = new ObjectName(this.name);
      mbeanServer.registerMBean(this, jmxName);
      Registry.bind (this.name, this);
   }

   public void start () throws Exception
   {
   }

   public void stop () throws Exception
   {
       // NR 200505:[JBCLUSTER-38] will be done at destroy instead
//      Registry.unbind (this.name);
//      ObjectName jmxName = new ObjectName(this.name);
//      mbeanServer.unregisterMBean (jmxName);
   }

    // NR 200505 : [JBCLUSTER-38] unbind DS MBean
   public void destroy () throws Exception
   {
      Registry.unbind (this.name);
      ObjectName jmxName = new ObjectName(this.name);
      mbeanServer.unregisterMBean (jmxName);
      
      partition.unsubscribeFromStateTransferEvents (SERVICE_NAME, this);
      partition.unregisterRPCHandler(SERVICE_NAME, this);
   }

   public String listContent () throws Exception
   {
      StringBuffer result = new StringBuffer ();
      Collection cats = this.getAllCategories ();
      Iterator catsIter = cats.iterator ();
      while (catsIter.hasNext ())
      {
         String category = (String)catsIter.next ();
         Iterator keysIter = this.getAllKeys(category).iterator ();

         result.append ("-----------------------------------------------\n");
         result.append ("Category : ").append (category).append ("\n\n");
         result.append ("KEY\t:\tVALUE\n");

         while (keysIter.hasNext ())
         {
            Serializable key = (Serializable) keysIter.next ();
            String value = this.get (category, key).toString ();
            result.append ("'").append(key);
            result.append ("'\t:\t'");
            result.append (value);
            result.append("'\n");
         }
         result.append ("\n");
      }
      return result.toString ();
   }

   public String listXmlContent () throws Exception
   {
      StringBuffer result = new StringBuffer ();
      Collection cats = this.getAllCategories ();
      Iterator catsIter = cats.iterator ();

      result.append ("<DistributedState>\n");

      while (catsIter.hasNext ())
      {
         String category = (String)catsIter.next ();
         Iterator keysIter = this.getAllKeys(category).iterator ();

         result.append ("\t<Category>\n");
         result.append ("\t\t<CategoryName>").append (category).append ("</CategoryName>\n");

         while (keysIter.hasNext ())
         {
            Serializable key = (Serializable) keysIter.next ();
            String value = this.get (category, key).toString ();
            result.append ("\t\t<Entry>\n");
            result.append ("\t\t\t<Key>").append (key).append ("</Key>\n");
            result.append ("\t\t\t<Value>").append (value).append ("</Value>\n");
            result.append ("\t\t</Entry>\n");
         }
         result.append ("\t</Category>\n");
      }
      result.append ("</DistributedState>\n");

      return result.toString ();
   }

   // DistributedState implementation ----------------------------------------------

   public void set (String category, Serializable key, Serializable value)
      throws Exception
   {
      set (category, key, value, true);
   }
   public void set (String category, Serializable key, Serializable value,
      boolean asynchronousCall) throws Exception
   {
      Object[] args = {category, key, value};
      if (asynchronousCall)
         partition.callAsynchMethodOnCluster (SERVICE_NAME, "_set", args, set_types, true);
      else
         partition.callMethodOnCluster (SERVICE_NAME, "_set", args, set_types, true);
      this._setInternal (category, key, value);
      notifyKeyListeners (category, key, value, true);
   }

   public Serializable remove (String category, Serializable key)
      throws Exception
   {
      return remove (category, key, true);
   }

   public Serializable remove (String category, Serializable key,
      boolean asynchronousCall) throws Exception
   {
      Object[] args = {category, key};
      if (asynchronousCall)
         partition.callAsynchMethodOnCluster (SERVICE_NAME, "_remove", args, remove_types, true);
      else
         partition.callMethodOnCluster (SERVICE_NAME, "_remove", args, remove_types, true);
      Serializable removed = this._removeInternal (category, key);
      notifyKeyListenersOfRemove (category, key, removed , true);
      return removed ;
   }

   public Serializable get (String category, Serializable key)
   {
      synchronized(this.categories)
      {
         HashMap cat = (HashMap)categories.get (category);
         if (cat == null) return null;

         return (Serializable)cat.get (key);
      }
   }

   public Collection getAllCategories ()
   {
      synchronized(this.categories)
      {
         return Collections.unmodifiableCollection(categories.keySet ());
      }
   }

   public Collection getAllKeys (String category)
   {
      synchronized(this.categories)
      {
         HashMap cat = (HashMap)categories.get (category);
         if (cat == null) return null;

         return Collections.unmodifiableCollection(cat.keySet ());
      }
   }

   public Collection getAllValues (String category)
   {
      synchronized(this.categories)
      {
         HashMap cat = (HashMap)categories.get (category);
         if (cat == null) return null;

         return Collections.unmodifiableCollection(cat.values ());
      }
   }


   public void registerDSListenerEx (String category, DSListenerEx subscriber)
   {
      registerListener(category, subscriber);
   }
   public void unregisterDSListenerEx (String category, DSListenerEx subscriber)
   {
      unregisterListener(category, subscriber);
   }
   public void registerDSListener (String category, DSListener subscriber)
   {
      registerListener(category, subscriber);
   }
   public void unregisterDSListener (String category, DSListener subscriber)
   {
      unregisterListener(category, subscriber);
   }

   // HAPartition RPC method invocations implementation ----------------------------------------------

   public void _set (String category, String key, Serializable value)
      throws Exception
   {
      this._setInternal (category, key, value);
      notifyKeyListeners (category, key, value, false);
   }
   public void _set (String category, Serializable key, Serializable value)
      throws Exception
   {
      this._setInternal (category, key, value);
      notifyKeyListeners (category, key, value, false);
   }

   public void _setInternal (String category, Serializable key, Serializable value)
      throws Exception
   {
      synchronized(this.categories)
      {
         HashMap cat = (HashMap)categories.get (category);
         if (cat == null)
         {
            cat = new HashMap ();
            categories.put (category, cat);
         }
         cat.put (key, value);
      }
   }

   public void _remove (String category, String key) throws Exception
   {
      Serializable removed = this._removeInternal (category, key);
      notifyKeyListenersOfRemove (category, key, removed, false);
   }
   public void _remove (String category, Serializable key) throws Exception
   {
      Serializable removed = this._removeInternal (category, key);
      notifyKeyListenersOfRemove (category, key, removed, false);
   }

   public Serializable _removeInternal (String category, Serializable key)
      throws Exception
   {
      synchronized(this.categories)
      {
         HashMap cat = (HashMap)categories.get (category);
         if (cat == null) return null;
         Object removed = cat.remove (key);
         if (removed != null)
         {
            if (cat.size () == 0)
            {
               categories.remove (category);
            }
         }
         return (Serializable)removed;
      }
   }

   // HAPartitionStateTransfer implementation ----------------------------------------------

   public Serializable getCurrentState ()
   {
      HashMap retval=new HashMap();
      Map.Entry entry;
      String catName;
      HashMap value, newVal;

      synchronized(this.categories)
      {

         for(Iterator it=this.categories.entrySet().iterator(); it.hasNext();)
         {
            entry=(Map.Entry)it.next(); // key: category name, value: HashMap
            catName=(String)entry.getKey();
            value=(HashMap)entry.getValue();
            newVal=value != null? (HashMap)value.clone() : null;
            retval.put(catName, newVal);
         }

         return retval;
      }
   }

   public void setCurrentState (Serializable newState)
   {
      synchronized (this.categories)
      {
         categories.clear ();
         categories.putAll ((HashMap)newState);
         if (keyListeners.size () > 0)
         {
            cleanupKeyListeners ();
         }
      }
   }

   // Package protected ---------------------------------------------

   // Protected -----------------------------------------------------

   protected void registerListener (String category, Object subscriber)
   {
      synchronized(this.keyListeners)
      {
         ArrayList listeners = (ArrayList)keyListeners.get (category);
         if (listeners == null)
         {
            listeners = new ArrayList ();
            keyListeners.put (category, listeners);
         }
         listeners.add (subscriber);
      }
   }

   protected void unregisterListener (String category, Object subscriber)
   {
      synchronized(this.keyListeners)
      {
         ArrayList listeners = (ArrayList)keyListeners.get (category);
         if (listeners == null) return;

         listeners.remove (subscriber);
         if (listeners.size () == 0)
         {
            keyListeners.remove (category);
         }
      }
   }

   protected void notifyKeyListeners (String category, Serializable key,
      Serializable value, boolean locallyModified)
   {
      synchronized(this.keyListeners)
      {
         ArrayList listeners = (ArrayList)keyListeners.get (category);
         if (listeners == null)
            return;
         String strKey = key.toString();

         for (int i = 0; i < listeners.size (); i++)
         {
            Object listener = listeners.get (i);
            if( listener instanceof DSListener )
            {
               DSListener dslistener = (DSListener) listener;
               dslistener.valueHasChanged (category, strKey, value, locallyModified);
            }
            else
            {
               DSListenerEx dslistener = (DSListenerEx) listener;
               dslistener.valueHasChanged (category, key, value, locallyModified);
            }
         }
      }
   }

   protected void notifyKeyListenersOfRemove (String category, Serializable key,
      Serializable oldContent, boolean locallyModified)
   {
      synchronized(this.keyListeners)
      {
         ArrayList listeners = (ArrayList)keyListeners.get (category);
         if (listeners == null)
            return;
         String strKey = key.toString();

         for (int i = 0; i < listeners.size (); i++)
         {
            Object listener = listeners.get (i);
            if( listener instanceof DSListener )
            {
               DSListener dslistener = (DSListener) listener;
               dslistener.keyHasBeenRemoved (category, strKey, oldContent, locallyModified);
            }
            else
            {
               DSListenerEx dslistener = (DSListenerEx) listener;
               dslistener.keyHasBeenRemoved (category, key, oldContent, locallyModified);
            }
         }
      }
   }

   protected void cleanupKeyListeners ()
   {
      // NOT IMPLEMENTED YET
   }

   // Private -------------------------------------------------------

   // Inner classes -------------------------------------------------

}
