/*
 * 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.framework.server;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Vector;

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 javax.management.MBeanServer;

import org.jgroups.JChannel;
import org.jgroups.MergeView;
import org.jgroups.View;
import org.jgroups.Message;
import org.jgroups.blocks.GroupRequest;
import org.jgroups.blocks.MethodCall;
import org.jgroups.stack.IpAddress;
import org.jgroups.util.Rsp;
import org.jgroups.util.RspList;

import org.jboss.invocation.MarshalledValueInputStream;
import org.jboss.invocation.MarshalledValueOutputStream;
import org.jboss.ha.framework.interfaces.DistributedReplicantManager;
import org.jboss.ha.framework.interfaces.DistributedState;
import org.jboss.ha.framework.interfaces.HAPartition;
import org.jboss.ha.framework.interfaces.ClusterNode;

import org.jboss.naming.NonSerializableFactory;
import org.jboss.logging.Logger;

/**
 * This class is an abstraction class for a JGroups RPCDispatch and JChannel.
 * It is a default implementation of HAPartition for the
 * <a href="http://www.jgroups.com/">JGroups</A> framework
 *
 * @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: 62255 $
 */
public class HAPartitionImpl
   extends org.jgroups.blocks.RpcDispatcher
   implements org.jgroups.MessageListener, org.jgroups.MembershipListener,
      HAPartition, AsynchEventHandler.AsynchEventProcessor
{
   private static class NoHandlerForRPC implements Serializable
   {
      static final long serialVersionUID = -1263095408483622838L;
   }

   // Constants -----------------------------------------------------

   // final MethodLookup method_lookup_clos = new MethodLookupClos();

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

   protected HashMap rpcHandlers = new HashMap();
   protected HashMap stateHandlers = new HashMap();
   /** Do we send any membership change notifications synchronously? */
   protected boolean allowSyncListeners = false;
   /** The synch HAMembershipListener and HAMembershipExtendedListeners */
   protected ArrayList synchListeners = new ArrayList();
   /** The asynch HAMembershipListener and HAMembershipExtendedListeners */
   protected ArrayList asynchListeners = new ArrayList();
   /** The handler used to send membership change notifications asynchronously */
   protected AsynchEventHandler asynchHandler;
   /** The current cluster partition members */
   protected Vector members = null;
   protected Vector jgmembers = null;

   public Vector history = null;

   /** The partition members other than this node */
   protected Vector otherMembers = null;
   protected Vector jgotherMembers = null;
   /** The JChannel name */
   protected String partitionName;
   /** the local JG IP Address */
   protected org.jgroups.stack.IpAddress localJGAddress = null;
   /** The cluster transport protocol address string */
   protected String nodeName;
   /** me as a ClusterNode */
   protected ClusterNode me = null;
   /** The timeout for cluster RPC calls */
   protected long timeout = 60000;
   /** The JGroups partition channel */
   protected JChannel channel;
   /** The cluster replicant manager */
   protected DistributedReplicantManagerImpl replicantManager;
   /** The cluster state manager */
   protected DistributedStateImpl dsManager;
   /** The cluster instance log category */
   protected Logger log;
   protected Logger clusterLifeCycleLog;   
   /** The current cluster view id */
   protected long currentViewId = -1;
   /** The JMX MBeanServer to use for registrations */
   protected MBeanServer server;
   /** Number of ms to wait for state */
   protected long state_transfer_timeout=60000;
   /** Whether to bind the partition into JNDI */
   protected boolean bindIntoJndi = true;

   /**
    * True if state was initialized during start-up.
    */
   protected boolean isStateSet = false;

   /**
    * An exception occuring upon fetch state.
    */
   protected Exception setStateException;
   private final Object stateLock = new Object();

   // Static --------------------------------------------------------
   
   /**
    * Creates an object from a byte buffer
    */
   public static Object objectFromByteBuffer (byte[] buffer) throws Exception
   {
      if(buffer == null) 
         return null;

      ByteArrayInputStream bais = new ByteArrayInputStream(buffer);
      MarshalledValueInputStream mvis = new MarshalledValueInputStream(bais);
      return mvis.readObject();
   }
   
   /**
    * Serializes an object into a byte buffer.
    * The object has to implement interface Serializable or Externalizable
    */
   public static byte[] objectToByteBuffer (Object obj) throws Exception
   {
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      MarshalledValueOutputStream mvos = new MarshalledValueOutputStream(baos);
      mvos.writeObject(obj);
      mvos.flush();
      return baos.toByteArray();
   }

   public long getStateTransferTimeout() {
      return state_transfer_timeout;
   }

   public void setStateTransferTimeout(long state_transfer_timeout) {
      this.state_transfer_timeout=state_transfer_timeout;
   }


   public long getMethodCallTimeout() {
      return timeout;
   }

   public void setMethodCallTimeout(long timeout) {
      this.timeout=timeout;
   }

    // Constructors --------------------------------------------------
       
   public HAPartitionImpl(String partitionName, org.jgroups.JChannel channel, boolean deadlock_detection, MBeanServer server) throws Exception
   {
      this(partitionName, channel, deadlock_detection);
      this.server = server;
   }
   
   public HAPartitionImpl(String partitionName, org.jgroups.JChannel channel, boolean deadlock_detection) throws Exception
   {
      super(channel, null, null, new Object(), deadlock_detection); // init RpcDispatcher with a fake target object
      this.log = Logger.getLogger(HAPartition.class.getName() + "." + partitionName);
      this.clusterLifeCycleLog = Logger.getLogger(HAPartition.class.getName() + ".lifecycle." + partitionName);
      this.channel = channel;
      this.partitionName = partitionName;
      this.history = new Vector();
      this.setMarshaller(new MarshallerImpl());
      logHistory ("Partition object created");
   }
   
    // Public --------------------------------------------------------
   
   public void init() throws Exception
   {
      log.info("Initializing");
      logHistory ("Initializing partition");

      // Subscribe to dHA events comming generated by the org.jgroups. protocol stack
      //
      log.debug("setMembershipListener");
      setMembershipListener(this);
      log.debug("setMessageListener");
      setMessageListener(this);
      
      // Create the DRM and link it to this HAPartition
      //
      log.debug("create replicant manager");
      this.replicantManager = new DistributedReplicantManagerImpl(this, this.server);
      log.debug("init replicant manager");
      this.replicantManager.init();
      log.debug("bind replicant manager");
      
      // Create the DS and link it to this HAPartition
      //
      log.debug("create distributed state");
      this.dsManager = new DistributedStateImpl(this, this.server);
      log.debug("init distributed state service");
      this.dsManager.init();
      log.debug("bind distributed state service");

      // Create the asynchronous handler for view changes
      asynchHandler = new AsynchEventHandler(this, "AsynchViewChangeHandler");
      
      log.debug("done initing.");
   }
   
   public void startPartition() throws Exception
   {
      // get current JG group properties
      //
      logHistory ("Starting partition");
      log.debug("get nodeName");
      this.localJGAddress = (IpAddress)channel.getLocalAddress();
      this.me = new ClusterNode(this.localJGAddress);
      this.nodeName = this.me.getName();

      log.debug("Get current members");
      View view = channel.getView();
      this.jgmembers = (Vector)view.getMembers().clone();
      this.members = translateAddresses(this.jgmembers); // TRANSLATE
      log.info("Number of cluster members: " + members.size());
      for(int m = 0; m > members.size(); m ++)
      {
         Object node = members.get(m);
         log.debug(node);
      }
      // Keep a list of other members only for "exclude-self" RPC calls
      //
      this.jgotherMembers = (Vector)view.getMembers().clone();
      this.jgotherMembers.remove (channel.getLocalAddress());
      this.otherMembers = translateAddresses(this.jgotherMembers); // TRANSLATE
      log.info ("Other members: " + this.otherMembers.size ());

      verifyNodeIsUnique (view.getMembers());

      // Update the initial view id
      //
      this.currentViewId = view.getVid().getId();

      // We must now synchronize new state transfer subscriber
      //
      fetchState();
      
      // We start now able to start our DRM and DS
      //
      this.replicantManager.start();
      this.dsManager.start();

      // Start the asynch listener handler thread
      asynchHandler.start();
      
      // Bind ourself in the public JNDI space if configured to do so
      if (!bindIntoJndi)
    	  return;
      
      Context ctx = new InitialContext();
      this.bind("/HAPartition/" + partitionName, this, HAPartitionImpl.class, ctx);
   }


   protected void fetchState() throws Exception
   {
      log.info("Fetching state (will wait for " + this.state_transfer_timeout + " milliseconds):");
      long start, stop;
      isStateSet = false;
      start = System.currentTimeMillis();
      boolean rc = channel.getState(null, this.state_transfer_timeout);
      if (rc)
      {
         synchronized (stateLock)
         {
            while (!isStateSet)
            {
               if (setStateException != null)
                  throw setStateException;

               try
               {
                  stateLock.wait();
               }
               catch (InterruptedException iex)
               {
               }
            }
         }
         stop = System.currentTimeMillis();
         log.info("state was retrieved successfully (in " + (stop - start) + " milliseconds)");
      }
      else
      {
         // No one provided us with state.
         // We need to find out if we are the coordinator, so we must
         // block until viewAccepted() is called at least once

         synchronized (members)
         {
            while (members.size() == 0)
            {
               log.debug("waiting on viewAccepted()");
               try
               {
                  members.wait();
               }
               catch (InterruptedException iex)
               {
               }
            }
         }

         if (isCurrentNodeCoordinator())
         {
            log.info("State could not be retrieved (we are the first member in group)");
         }
         else
         {
            throw new IllegalStateException("Initial state transfer failed: " +
               "Channel.getState() returned false");
         }
      }
   }

   public void closePartition() throws Exception
   {
      logHistory ("Closing partition");
      log.info("Closing partition " + partitionName);

      try
      {
         asynchHandler.stop();
      }
      catch( Exception e)
      {
         log.warn("Failed to stop asynchHandler", e);
      }

      // Stop the DRM and DS services
      //
      try
      {
         this.replicantManager.stop();
      }
      catch (Exception e)
      {
         log.error("operation failed", e);
      }

      try
      {
         this.dsManager.stop();
      }
      catch (Exception e)
      {
         log.error("operation failed", e);
      }

//    NR 200505 : [JBCLUSTER-38] replace channel.close() by a disconnect and
//    add the destroyPartition() step
      try
      {
//          channel.close();
          channel.disconnect();
      }
      catch (Exception e)
      {
         log.error("operation failed", e);
      }

      if (bindIntoJndi)
      {
         String boundName = "/HAPartition/" + partitionName;
         InitialContext ctx = new InitialContext();
         try
         {
            ctx.unbind(boundName);
         }
         finally
         {
            ctx.close();
         }
         NonSerializableFactory.unbind (boundName);    	  
      }
      
      log.info("Partition " + partitionName + " closed.");
   }

// NR 200505 : [JBCLUSTER-38] destroy partition close the channel
   public void destroyPartition()  throws Exception
   {

      try
      {
         this.replicantManager.destroy();
      }
      catch (Exception e)
      {
         log.error("operation failed", e);
      }

      try
      {
         this.dsManager.destroy();
      }
      catch (Exception e)
      {
         log.error("operation failed", e);
      }
      try
      {
         channel.close();
      }
      catch (Exception e)
      {
         log.error("operation failed", e);
      }

      log.info("Partition " + partitionName + " destroyed.");
  }
   // org.jgroups.MessageListener implementation ----------------------------------------------

   // MessageListener methods
   //
   public byte[] getState()
   {
      logHistory ("getState called on partition");
      boolean debug = log.isDebugEnabled();
      
      log.debug("getState called.");
      try
      {
         // we now get the sub-state of each HAPartitionStateTransfer subscribers and
         // build a "macro" state
         //
         HashMap state = new HashMap();
         Iterator keys = stateHandlers.keySet().iterator();
         while (keys.hasNext())
         {
            String key = (String)keys.next();
            HAPartition.HAPartitionStateTransfer subscriber = (HAPartition.HAPartitionStateTransfer)stateHandlers.get(key);
            if (debug)
               log.debug("getState for " + key);
            state.put(key, subscriber.getCurrentState());
         }
         return objectToByteBuffer(state);
      }
      catch (Exception ex)
      {
         log.error("getState failed", ex);
      }
      return null;
   }
   
   public void setState(byte[] obj)
   {
      logHistory ("setState called on partition");
      try
      {
         log.debug("setState called");
         if (obj == null)
         {
            log.debug("state is null");
            return;
         }
         
         long used_mem_before, used_mem_after;
         int state_size=obj != null? obj.length : 0;
         Runtime rt=Runtime.getRuntime();
         used_mem_before=rt.totalMemory() - rt.freeMemory();

         HashMap state = (HashMap)objectFromByteBuffer(obj);
         java.util.Iterator keys = state.keySet().iterator();
         while (keys.hasNext())
         {
            String key = (String)keys.next();
            log.debug("setState for " + key);
            Object someState = state.get(key);
            HAPartition.HAPartitionStateTransfer subscriber = (HAPartition.HAPartitionStateTransfer)stateHandlers.get(key);
            if (subscriber != null)
            {
               try
               {
                  subscriber.setCurrentState((java.io.Serializable)someState);
               }
               catch (Exception e)
               {
                  // Don't let issues with one subscriber affect others
                  // unless it is DRM or DS, which are really internal 
                  // functions of the HAPartition
                  if (DistributedReplicantManagerImpl.SERVICE_NAME.equals(key) 
                        || DistributedStateImpl.SERVICE_NAME.equals(key))
                  {
                     if (e instanceof RuntimeException)
                        throw (RuntimeException) e;
                     else
                        throw new RuntimeException(e);
                  }
                  else
                  {
                     log.error("Caught exception setting state to " + subscriber, e);
                  }
               }
            }
            else
            {
               log.debug("There is no stateHandler for: " + key);
            }
         }

         used_mem_after=rt.totalMemory() - rt.freeMemory();
         log.debug("received a state of " + state_size + " bytes; expanded memory by " +
               (used_mem_after - used_mem_before) + " bytes (used memory before: " + used_mem_before +
               ", used memory after: " + used_mem_after + ")");
         
         isStateSet = true;
      }
      catch (Throwable t)
      {
         log.error("failed setting state", t);
         if (t instanceof Exception)
            setStateException = (Exception) t;
         else
            setStateException = new Exception(t);
      }
      finally
      {
         synchronized (stateLock)
         {
            // Notify wait that state has been set.
            stateLock.notifyAll();
         }
      }
   }
   
   public void receive(org.jgroups.Message msg)
   { /* complete */}
   
   // org.jgroups.MembershipListener implementation ----------------------------------------------
   
   public void suspect(org.jgroups.Address suspected_mbr)
   {      
      logHistory ("Node suspected: " + (suspected_mbr==null?"null":suspected_mbr.toString()));
      if (isCurrentNodeCoordinator ())
         clusterLifeCycleLog.info ("Suspected member: " + suspected_mbr);
      else
         log.info("Suspected member: " + suspected_mbr);
   }

   public void block() {}
   
   /** Notification of a cluster view change. This is done from the JG protocol
    * handlder thread and we must be careful to not unduly block this thread.
    * Because of this there are two types of listeners, synchronous and
    * asynchronous. The synchronous listeners are messaged with the view change
    * event using the calling thread while the asynchronous listeners are
    * messaged using a seperate thread.
    *
    * @param newView
    */
   public void viewAccepted(View newView)
   {
      try
      {
         // we update the view id
         //
         this.currentViewId = newView.getVid().getId();

         // Keep a list of other members only for "exclude-self" RPC calls
         //
         this.jgotherMembers = (Vector)newView.getMembers().clone();
         this.jgotherMembers.remove (channel.getLocalAddress());
         this.otherMembers = translateAddresses (this.jgotherMembers); // TRANSLATE!
         Vector translatedNewView = translateAddresses ((Vector)newView.getMembers().clone());
         logHistory ("New view: " + translatedNewView + " with viewId: " + this.currentViewId +
                     " (old view: " + this.members + " )");


         // Save the previous view and make a copy of the new view
         Vector oldMembers = this.members;

         Vector newjgMembers = (Vector)newView.getMembers().clone();
         Vector newMembers = translateAddresses(newjgMembers); // TRANSLATE
         if (this.members == null)
         {
            // Initial viewAccepted
            this.members = newMembers;
            this.jgmembers = newjgMembers;
            log.debug("ViewAccepted: initial members set");
            return;
         }
         this.members = newMembers;
         this.jgmembers = newjgMembers;

         int difference = 0;
         if (oldMembers == null)
            difference = newMembers.size () - 1;
         else
            difference = newMembers.size () - oldMembers.size ();
         
         if (isCurrentNodeCoordinator ())
            clusterLifeCycleLog.info ("New cluster view for partition " + this.partitionName + " (id: " +
                                      this.currentViewId + ", delta: " + difference + ") : " + this.members);
         else
            log.info("New cluster view for partition " + this.partitionName + ": " +
                     this.currentViewId + " (" + this.members + " delta: " + difference + ")");

         // Build a ViewChangeEvent for the asynch listeners
         ViewChangeEvent event = new ViewChangeEvent();
         event.viewId = currentViewId;
         event.allMembers = translatedNewView;
         event.deadMembers = getDeadMembers(oldMembers, event.allMembers);
         event.newMembers = getNewMembers(oldMembers, event.allMembers);
         event.originatingGroups = null;
         // if the new view occurs because of a merge, we first inform listeners of the merge
         if(newView instanceof MergeView)
         {
            MergeView mergeView = (MergeView) newView;
            event.originatingGroups = mergeView.getSubgroups();
         }

         log.debug("membership changed from " + 
                   (oldMembers == null ? 0 : oldMembers.size()) + " to " + 
                   event.allMembers.size());
         // Put the view change to the asynch queue
         this.asynchHandler.queueEvent(event);

         // Broadcast the new view to the synchronous view change listeners
         if (this.allowSyncListeners)
         {
            this.notifyListeners(synchListeners, event.viewId, event.allMembers,
                  event.deadMembers, event.newMembers, event.originatingGroups);
         }
      }
      catch (Exception ex)
      {
         log.error("ViewAccepted failed", ex);
      }
   }

   // HAPartition implementation ----------------------------------------------
   
   public String getNodeName()
   {
      return nodeName;
   }
   
   public String getPartitionName()
   {
      return partitionName;
   }
   
   public DistributedReplicantManager getDistributedReplicantManager()
   {
      return replicantManager;
   }
   
   public DistributedState getDistributedStateService()
   {
      return this.dsManager;
   }

   public long getCurrentViewId()
   {
      return this.currentViewId;
   }
   
   public Vector getCurrentView()
   {
      Vector result = new Vector (this.members.size());
      for (int i = 0; i < members.size(); i++)
      {
         result.add( ((ClusterNode) members.elementAt(i)).getName() );
      }
      return result;
   }

   public ClusterNode[] getClusterNodes ()
   {
      ClusterNode[] nodes = new ClusterNode[this.members.size()];
      this.members.toArray(nodes);
      return nodes;
   }

   public ClusterNode getClusterNode ()
   {
      return me;
   }

   public boolean isCurrentNodeCoordinator ()
   {
      if(this.members == null || this.members.size() == 0 || this.me == null)
         return false;
     return this.members.elementAt (0).equals (this.me);
   }

   // ***************************
   // ***************************
   // RPC multicast communication
   // ***************************
   // ***************************
   //
   public void registerRPCHandler(String objName, Object subscriber)
   {
      rpcHandlers.put(objName, subscriber);
   }
   
   public void unregisterRPCHandler(String objName, Object subscriber)
   {
      rpcHandlers.remove(objName);
   }
      

   /**
    *
    * @param objName
    * @param methodName
    * @param args
    * @param excludeSelf
    * @return
    * @throws Exception
    * @deprecated Use {@link #callMethodOnCluster(String,String,Object[],Class[], boolean)} instead
    */
   public ArrayList callMethodOnCluster(String objName, String methodName,
      Object[] args, boolean excludeSelf) throws Exception
   {
      return callMethodOnCluster(objName, methodName, args, null, excludeSelf);
   }

   /**
    * This function is an abstraction of RpcDispatcher.
    */
   public ArrayList callMethodOnCluster(String objName, String methodName,
      Object[] args, Class[] types, boolean excludeSelf) throws Exception
   {
      return callMethodOnCluster(objName, methodName, args, types, excludeSelf, this.timeout);
   }


   public ArrayList callMethodOnCluster(String objName, String methodName,
       Object[] args, Class[] types, boolean excludeSelf, long methodTimeout) throws Exception
   {
      ArrayList rtn = new ArrayList();
      MethodCall m=null;
      RspList rsp = null;
      boolean trace = log.isTraceEnabled();

      if(types != null)
         m=new MethodCall(objName + "." + methodName, args, types);
      else
         m=new MethodCall(objName + "." + methodName, args);

      if (excludeSelf)
      {
         if( trace )
         {
            log.trace("callMethodOnCluster(true), objName="+objName
               +", methodName="+methodName+", members="+jgotherMembers);
         }
         rsp = this.callRemoteMethods(this.jgotherMembers, m, GroupRequest.GET_ALL, methodTimeout);
      }
      else
      {
         if( trace )
         {
            log.trace("callMethodOnCluster(false), objName="+objName
               +", methodName="+methodName+", members="+members);
         }
         rsp = this.callRemoteMethods(null, m, GroupRequest.GET_ALL, methodTimeout);
      }

      if (rsp != null)
      {
         for (int i = 0; i < rsp.size(); i++)
         {
            Object item = rsp.elementAt(i);
            if (item instanceof Rsp)
            {
               Rsp response = (Rsp) item;
               // Only include received responses
               boolean wasReceived = response.wasReceived();
               if( wasReceived == true )
               {
                  item = response.getValue();
                  if (!(item instanceof NoHandlerForRPC))
                     rtn.add(item);
               }
               else if( trace )
                  log.trace("Ignoring non-received response: "+response);
            }
            else
            {
               if (!(item instanceof NoHandlerForRPC))
                  rtn.add(item);
               else if( trace )
                  log.trace("Ignoring NoHandlerForRPC");
            }
         }
      }

      return rtn;
    }

   /**
    * Calls method on Cluster coordinator node only.  The cluster coordinator node is the first node to join the
    * cluster.
    * and is replaced
    * @param objName
    * @param methodName
    * @param args
    * @param types
    * @param excludeSelf
    * @return
    * @throws Exception
    */
   public ArrayList callMethodOnCoordinatorNode(String objName, String methodName,
          Object[] args, Class[] types,boolean excludeSelf) throws Exception
      {
      return callMethodOnCoordinatorNode(objName,methodName,args,types,excludeSelf,this.timeout);
      }

   /**
    * Calls method on Cluster coordinator node only.  The cluster coordinator node is the first node to join the
    * cluster.
    * and is replaced
    * @param objName
    * @param methodName
    * @param args
    * @param types
    * @param excludeSelf
    * @param methodTimeout
    * @return
    * @throws Exception
    */
   public ArrayList callMethodOnCoordinatorNode(String objName, String methodName,
          Object[] args, Class[] types,boolean excludeSelf, long methodTimeout) throws Exception
      {
         ArrayList rtn = new ArrayList();
         MethodCall m=null;
         RspList rsp = null;
         boolean trace = log.isTraceEnabled();

         if(types != null)
            m=new MethodCall(objName + "." + methodName, args, types);
         else
            m=new MethodCall(objName + "." + methodName, args);

         if( trace )
         {
            log.trace("callMethodOnCoordinatorNode(false), objName="+objName
               +", methodName="+methodName);
         }

         // the first cluster view member is the coordinator
         Vector coordinatorOnly = new Vector();
         // If we are the coordinator, only call ourself if 'excludeSelf' is false
         if (false == isCurrentNodeCoordinator () ||
             false == excludeSelf)
            coordinatorOnly.addElement(this.jgmembers.elementAt (0));

         rsp = this.callRemoteMethods(coordinatorOnly, m, GroupRequest.GET_ALL, methodTimeout);

         if (rsp != null)
         {
            for (int i = 0; i < rsp.size(); i++)
            {
               Object item = rsp.elementAt(i);
               if (item instanceof Rsp)
               {
                  Rsp response = (Rsp) item;
                  // Only include received responses
                  boolean wasReceived = response.wasReceived();
                  if( wasReceived == true )
                  {
                     item = response.getValue();
                     if (!(item instanceof NoHandlerForRPC))
                        rtn.add(item);
                  }
                  else if( trace )
                     log.trace("Ignoring non-received response: "+response);
               }
               else
               {
                  if (!(item instanceof NoHandlerForRPC))
                     rtn.add(item);
                  else if( trace )
                     log.trace("Ignoring NoHandlerForRPC");
               }
            }
         }

         return rtn;
       }


   /**
    *
    * @param objName
    * @param methodName
    * @param args
    * @param excludeSelf
    * @throws Exception
    * @deprecated Use {@link #callAsynchMethodOnCluster(String, String, Object[], Class[], boolean)} instead
    */
   public void callAsynchMethodOnCluster(String objName, String methodName,
      Object[] args, boolean excludeSelf) throws Exception {
      callAsynchMethodOnCluster(objName, methodName, args, null, excludeSelf);
   }





   /**
    * This function is an abstraction of RpcDispatcher for asynchronous messages
    */
   public void callAsynchMethodOnCluster(String objName, String methodName,
      Object[] args, Class[] types, boolean excludeSelf) throws Exception
   {
      MethodCall m = null;
      boolean trace = log.isTraceEnabled();

      if(types != null)
         m=new MethodCall(objName + "." + methodName, args, types);
      else
         m=new MethodCall(objName + "." + methodName, args);

      if (excludeSelf)
      {
         if( trace )
         {
            log.trace("callAsynchMethodOnCluster(true), objName="+objName
               +", methodName="+methodName+", members="+jgotherMembers);
         }
         this.callRemoteMethods(this.jgotherMembers, m, GroupRequest.GET_NONE, timeout);
      }
      else
      {
         if( trace )
         {
            log.trace("callAsynchMethodOnCluster(false), objName="+objName
               +", methodName="+methodName+", members="+members);
         }
         this.callRemoteMethods(null, m, GroupRequest.GET_NONE, timeout);
      }
   }
   
   // *************************
   // *************************
   // State transfer management
   // *************************
   // *************************
   //      
   public void subscribeToStateTransferEvents(String objectName, HAPartitionStateTransfer subscriber)
   {
      stateHandlers.put(objectName, subscriber);
   }
   
   public void unsubscribeFromStateTransferEvents(String objectName, HAPartitionStateTransfer subscriber)
   {
      stateHandlers.remove(objectName);
   }
   
   // *************************
   // *************************
   // Group Membership listeners
   // *************************
   // *************************
   //   
   public void registerMembershipListener(HAMembershipListener listener)
   {
      boolean isAsynch = (this.allowSyncListeners == false) 
            || (listener instanceof AsynchHAMembershipListener)
            || (listener instanceof AsynchHAMembershipExtendedListener);
      if( isAsynch ) {
         synchronized(this.asynchListeners) {
            this.asynchListeners.add(listener);
         }
      }
      else  { 
         synchronized(this.synchListeners) {
            this.synchListeners.add(listener);
         }
      }
   }
   
   public void unregisterMembershipListener(HAMembershipListener listener)
   {
      boolean isAsynch = (this.allowSyncListeners == false) 
            || (listener instanceof AsynchHAMembershipListener)
            || (listener instanceof AsynchHAMembershipExtendedListener);
      if( isAsynch ) {
         synchronized(this.asynchListeners) {
            this.asynchListeners.remove(listener);
         }
      }
      else  { 
         synchronized(this.synchListeners) {
            this.synchListeners.remove(listener);
         }
      }
   }
   
   public boolean getAllowSynchronousMembershipNotifications()
   {
      return allowSyncListeners;
   }

   public void setAllowSynchronousMembershipNotifications(boolean allowSync)
   {      
      this.allowSyncListeners = allowSync;
   }
   
   // org.jgroups.RpcDispatcher overrides ---------------------------------------------------
   
   /**
    * Message contains MethodCall. Execute it against *this* object and return result.
    * Use MethodCall.Invoke() to do this. Return result.
    *
    * This overrides RpcDispatcher.Handle so that we can dispatch to many different objects.
    * @param req The org.jgroups. representation of the method invocation
    * @return The serializable return value from the invocation
    */
   public Object handle(Message req)
   {
      Object body = null;
      Object retval = null;
      MethodCall  method_call = null;
      boolean trace = log.isTraceEnabled();
      
      if( trace )
         log.trace("Partition " + partitionName + " received msg");
      if(req == null || req.getBuffer() == null)
      {
         log.warn("message or message buffer is null !");
         return null;
      }
      
      try
      {
         body = objectFromByteBuffer(req.getBuffer());
      }
      catch(Exception e)
      {
         log.warn("failed unserializing message buffer (msg=" + req + ")", e);
         return null;
      }
      
      if(body == null || !(body instanceof MethodCall))
      {
         log.warn("message does not contain a MethodCall object !");
         return null;
      }
      
      // get method call informations
      //
      method_call = (MethodCall)body;
      String methodName = method_call.getName();      
      
      if( trace )
         log.trace("pre methodName: " + methodName);
      
      int idx = methodName.lastIndexOf('.');
      String handlerName = methodName.substring(0, idx);
      String newMethodName = methodName.substring(idx + 1);
      
      if( trace ) 
      {
         log.trace("handlerName: " + handlerName + " methodName: " + newMethodName);
         log.trace("Handle: " + methodName);
      }
      
      // prepare method call
      method_call.setName(newMethodName);
      Object handler = rpcHandlers.get(handlerName);
      if (handler == null)
      {
         if( trace )
            log.debug("No rpc handler registered under: "+handlerName);
         return new NoHandlerForRPC();
      }

      /* Invoke it and just return any exception with trace level logging of
      the exception. The exception semantics of a group rpc call are weak as
      the return value may be a normal return value or the exception thrown.
      */
      try
      {
         retval = method_call.invoke(handler);
         if( trace )
            log.trace("rpc call return value: "+retval);
      }
      catch (Throwable t)
      {
         if( trace )
            log.trace("rpc call threw exception", t);
         retval = t;
      }
      
      return retval;
   }
   
   // AsynchEventHandler.AsynchEventProcessor -----------------------

   public void processEvent(Object event)
   {
      ViewChangeEvent vce = (ViewChangeEvent) event;
      notifyListeners(asynchListeners, vce.viewId, vce.allMembers,
            vce.deadMembers, vce.newMembers, vce.originatingGroups);
      
   }
   
   
   // Package protected ---------------------------------------------
   
   // Protected -----------------------------------------------------

   protected void verifyNodeIsUnique (Vector javaGroupIpAddresses) throws Exception
   {
      byte[] localUniqueName = this.localJGAddress.getAdditionalData();
      if (localUniqueName == null)
      {
         log.warn("No additional information has been found in the JavaGroup address: " +
                  "make sure you are running with a correct version of JGroups and that the protocol " +
                  " you are using supports the 'additionalData' behaviour");
         return;
      }

      for (int i = 0; i < javaGroupIpAddresses.size(); i++)
      {
         IpAddress address = (IpAddress) javaGroupIpAddresses.elementAt(i);
         if (!address.equals(this.localJGAddress))
         {
            if (localUniqueName.equals(address.getAdditionalData()))
               throw new Exception ("Local node removed from cluster (" + this.localJGAddress + "): another node (" + address + ") publicizing the same name was already there");
         }
      }
   }

   /**
    * Helper method that binds the partition in the JNDI tree.
    * @param jndiName Name under which the object must be bound
    * @param who Object to bind in JNDI
    * @param classType Class type under which should appear the bound object
    * @param ctx Naming context under which we bind the object
    * @throws Exception Thrown if a naming exception occurs during binding
    */   
   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.rebind (n.get (0), ref);
   }
   
   /**
    * Helper method that returns a vector of dead members from two input vectors: new and old vectors of two views.
    * Dead members are old - new members.
    * @param oldMembers Vector of old members
    * @param newMembers Vector of new members
    * @return Vector of members that have died between the two views, can be empty.
    */   
   protected Vector getDeadMembers(Vector oldMembers, Vector newMembers)
   {
      boolean debug = log.isDebugEnabled();
      if(oldMembers == null) oldMembers=new Vector();
      if(newMembers == null) newMembers=new Vector();
      Vector dead=(Vector)oldMembers.clone();
      dead.removeAll(newMembers);
      if(dead.size() > 0 && debug)
         log.debug("dead members: " + dead);
      return dead;
   }
   
   /**
    * Helper method that returns a vector of new members from two input vectors: new and old vectors of two views.
    * @param oldMembers Vector of old members
    * @param allMembers Vector of new members
    * @return Vector of members that have joined the partition between the two views
    */   
   protected Vector getNewMembers(Vector oldMembers, Vector allMembers)
   {
      if(oldMembers == null) oldMembers=new Vector();
      if(allMembers == null) allMembers=new Vector();
      Vector newMembers=(Vector)allMembers.clone();
      newMembers.removeAll(oldMembers);
      return newMembers;
   }

   protected void notifyListeners(ArrayList theListeners, long viewID,
      Vector allMembers, Vector deadMembers, Vector newMembers,
      Vector originatingGroups)
   {
      log.debug("Begin notifyListeners, viewID: "+viewID);
      
      synchronized(theListeners)
      {
         // JBAS-3619 -- don't hold synch lock while notifying
         theListeners = (ArrayList) theListeners.clone();
      }
      
      for (int i = 0; i < theListeners.size(); i++)
      {
         HAMembershipListener aListener = null;
         try
         {
            aListener = (HAMembershipListener) theListeners.get(i);
            if(originatingGroups != null && (aListener instanceof HAMembershipExtendedListener))
            {
               HAMembershipExtendedListener exListener = (HAMembershipExtendedListener) aListener;
               exListener.membershipChangedDuringMerge (deadMembers, newMembers,
                  allMembers, originatingGroups);
            }
            else
            {
               aListener.membershipChanged(deadMembers, newMembers, allMembers);
            }
         }
         catch (Throwable e)
         {
            // a problem in a listener should not prevent other members to receive the new view
            log.warn("HAMembershipListener callback failure: "+aListener, e);
         }
      }
      log.debug("End notifyListeners, viewID: "+viewID);
   }
   
   /*
    * Allows caller to specify whether the partition instance should be bound into JNDI.  Default value is true.
    * This method must be called before the partition is started as the binding occurs during startup.
    * 
    * @param bind  Whether to bind the partition into JNDI.
    */
   public void setBindIntoJndi(boolean bind)
   {
	   bindIntoJndi = bind;
   }
   
   /*
    * Allows caller to determine whether the partition instance has been bound into JNDI.
    * 
    * @return true if the partition has been bound into JNDI.
    */
   public boolean getBindIntoJndi()
   {
	   return bindIntoJndi;
   }
   
   protected Vector translateAddresses (Vector jgAddresses)
   {
      if (jgAddresses == null)
         return null;

      Vector result = new Vector (jgAddresses.size());
      for (int i = 0; i < jgAddresses.size(); i++)
      {
         IpAddress addr = (IpAddress) jgAddresses.elementAt(i);
         result.add(new ClusterNode (addr));
      }

      return result;
   }

   public void logHistory (String message)
   {
      try
      {
         history.add(new SimpleDateFormat().format (new Date()) + " : " + message);
      }
      catch (Exception ignored){}
   }

   /** A simple data class containing the view change event needed to
    * message the HAMembershipListeners
    */
   private static class ViewChangeEvent
   {
      long viewId;
      Vector deadMembers;
      Vector newMembers;
      Vector allMembers;
      Vector originatingGroups;
   }
   
   private class MarshallerImpl implements org.jgroups.blocks.RpcDispatcher.Marshaller
   {

      public Object objectFromByteBuffer(byte[] buf) throws Exception
      {
         return HAPartitionImpl.objectFromByteBuffer(buf);
      }

      public byte[] objectToByteBuffer(Object obj) throws Exception
      {
         return HAPartitionImpl.objectToByteBuffer(obj);
      }
      
   }

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

}