/*
    * Copyright (c) 1998-2002 The Jgroup Team.
    *
    * This program is free software; you can redistribute it and/or modify
    * it under the terms of the GNU Lesser General Public License version 2 as
    * published by the Free Software Foundation.
    *
    * This program 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 program; if not, write to the Free Software
   * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
   *
   */
  
  package jgroup.relacs.gm;
  
  import static jgroup.util.log.ViewEvent.Type.Merge;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.util.ArrayList;
  import java.util.List;
  
  import javax.naming.ConfigurationException;
  
  import jgroup.core.JgroupException;
  import jgroup.core.MemberId;
  import jgroup.core.MemberTable;
  import jgroup.core.MembershipListener;
  import jgroup.core.MembershipService;
  import jgroup.core.MergingListener;
  import jgroup.core.MergingService;
  import jgroup.core.View;
  import jgroup.core.multicast.MulticastListener;
  import jgroup.core.multicast.MulticastService;
  import jgroup.relacs.rmi.MarshalInputStream;
  import jgroup.relacs.rmi.MarshalOutputStream;
  import jgroup.util.log.Eventlogger;
  import jgroup.util.log.ViewEvent;
  
  import org.apache.log4j.Logger;
  
  /**
   *  This <code>MergingLayer</code> class implements the Jgroup state
   *  merging service.  It intercepts and delays view change events so
   *  that the reconciliation protocol can complete before passing the
   *  view change on to application member.
   *
   *  @author Alberto Montresor
   *  @author Hein Meling
   *  @since 0.7
   */
  public class MergingLayer
    implements MergingService, MembershipService, MembershipListener, MulticastListener
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(MergingLayer.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Constants
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Protocol name used to distinguish messages from other protocols. */
    private static final String PROTOCOL_NAME = "SMS";
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /* Underlying membership service */
    private final MembershipService membershipService;
  
    /* Underlying multicast service */
    private final MulticastService multicastService;
  
    /**
     * Dynamic list of membership listeners.  Note that the ordering of
     * these listeners are important.
     */
    private final List<MembershipListener> membershipListeners =
      new ArrayList<MembershipListener>();
  
    /**
     * Dynamic list of merging listeners.  Note that the ordering of
     * these listeners are important.
     */
    private final List<MergingListener> mergingListeners =
     new ArrayList<MergingListener>();
 
   /** Member table used by this layer (from MembershipLayer) */
   private final MemberTable table;
 
   /**
    * Member table provided by this layer
    * This is to hide details stored in 'table' by this layer.
    */
   private final MemberTable providedTable = new MemberTable();
 
   /** The MemberId of this member */
   private final MemberId me;
 
   /** The most recent view installed (not updated until the end of viewChange()) */
   private View view;
 
   /** The composition of the last view */
   private MemberId[] members;
 
   /** The number of remaining put state messages before merging is complete */
   private int remainingPutStates;
 
   /** The previous view */
   private View previousView;
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructors
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *
    * @exception ConfigurationException
    *            raised when the distributed system configuration object contains an error
    * @exception IOException
    *            if an I/O error occurs when creating communication sockets.
    */
   private MergingLayer(MembershipService membershipService, MulticastService multicastService)
     throws JgroupException
   {
     this.membershipService = membershipService;
     this.multicastService = multicastService;
 
     table = membershipService.getMemberTable();
     me = membershipService.getMyIdentifier();
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Static factory
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   public static MergingLayer getLayer(MembershipService membService, MulticastService mcastService)
     throws JgroupException
   {
     return new MergingLayer(membService, mcastService);
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Layer interface methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Add a listener object for this layer to provide upcalls to,
    *  in response to membership and merge events.  The same listener
    *  object may listen to both types of events or just one of them.
    */
   public void addListener(Object listener)
   {
     if (listener instanceof MembershipListener && !membershipListeners.contains(listener))
       membershipListeners.add((MembershipListener) listener);
     if (listener instanceof MergingListener && !mergingListeners.contains(listener))
       mergingListeners.add((MergingListener) listener);
     if (!(listener instanceof MergingListener) && !(listener instanceof MembershipListener))
       throw new IllegalArgumentException("Specified listener does not implement a MembershipListener or MergingListener");
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // FinalizeLayer interface methods (invoked to complete the group manager construction)
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Post initialization of the <code>MembershipLayer</code>; invoked once
    *  all group manager layers have been constructed.  This method is defined
    *  in the <code>FinalizeLayer</code> inherited through the
    *  <code>MembershipSerivce</code>.
    */
   public void complete(Object server)
     throws JgroupException
   {
     membershipService.complete(server);
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods of MembershipService
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#join(int)
    */
   public void join(int gid)
     throws JgroupException
   {
     membershipService.join(gid);
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#join()
    */
   public void join()
     throws JgroupException
   {
     membershipService.join();
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#leave()
    */
   public void leave()
     throws JgroupException
   {
     membershipService.leave();
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#isMember()
    */
   public boolean isMember()
   {
     return membershipService.isMember();
   }
   
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#isJoining()
    */
   public boolean isJoining()
   {
     return membershipService.isJoining();
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#isMemberOrJoining()
    */
   public boolean isMemberOrJoining()
   {
     return membershipService.isMemberOrJoining();
   }
   
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#isLeader()
    */
   public boolean isLeader()
   {
     return membershipService.isLeader();
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#getMyIdentifier()
    */
   public MemberId getMyIdentifier()
   {
     return me;
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#getGid()
    */
   public int getGid()
   {
     return membershipService.getGid();
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#getMemberTable()
    */
   public MemberTable getMemberTable()
   {
     return providedTable;
   }
 
   public int members()
   {
     return membershipService.members();
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#getViewIndex()
    */
   public int getViewIndex()
   {
     return membershipService.getViewIndex();
   }
 
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipService#getMemberIndex()
    */
   public int getMemberIndex()
   {
     return membershipService.getMemberIndex();
   }
 
   
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from MembershipListener
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipListener#viewChange(jgroup.core.View)
    */
   @AllowDuplicateViews public void viewChange(final View view)
   {
     if (log.isDebugEnabled())
       log.debug("MergingLayer: viewChange");
     // Update the member table associated to this group
     providedTable.viewChange(view);
     // Store the current view for use in deliverStream
     this.view = view;
     remainingPutStates = view.mergingViews();
     if (log.isDebugEnabled())
       log.debug("REMAINING PUTSTATES: " + remainingPutStates);
 
     /*
      * Associate new (empty) information with new members, and also
      * associate new information with members that have recovered.
      */
     members = view.getMembers();
     for (int i=0; i < members.length; i++) {
       int status = table.getStatus(members[i]);
       if (status == MemberTable.NEWMEMBER || status == MemberTable.RECOVERING)
         table.put(members[i], new MemberInfo(members[i], members[i].equals(me)));
     }
 
     /*
      * Get the total list of members registered in table, including those in the
      * most recent view, and those that have recently partitioned or crashed.
      */
     Object[] objs = table.elements();
     MemberInfo[] mdata = new MemberInfo[objs.length];
     for (int i=0; i < objs.length; i++) 
       mdata[i] = (MemberInfo) objs[i];
 
     /*
      * Update the mset and lset values, and compute if an update message
      * is necessary.
      */
     boolean needupdate = false;
     for (int i=0; i < mdata.length; i++) {
       if (!table.isMember(mdata[i].id)) {
         mdata[i].mset = false;
         mdata[i].lset = false;
         mdata[i].tset = false;
       } else {
         mdata[i].tset = true;
         if (!mdata[i].mset && !mdata[i].id.equals(me))
           needupdate = true;
       }
       if (log.isDebugEnabled()) {
         StringBuilder buf = new StringBuilder();
         buf.append(mdata[i]);
         buf.append(", needupdate=");
         buf.append(needupdate);
         log.debug(buf.toString());
       }
     }
 
     if (needupdate) {
       if (Eventlogger.ENABLED)
         Eventlogger.logEventFlush(new ViewEvent(Merge, view));
       if (hasLocalCoordinator(mdata)) {
         if (log.isDebugEnabled())
           log.debug("I'm merging coordinator");
         try {
           OutputStream stream = multicastService.getMessage(PROTOCOL_NAME);
           MarshalOutputStream msg = new MarshalOutputStream(stream);
           int[] owners = getOwners(mdata);
           int[] dests = getDests(mdata);
           MemberId[] destIds = new MemberId[dests.length];
           for (int j = 0; j < dests.length; j++)
             destIds[j] = members[dests[j]];
 
           msg.writeInt(owners.length);
           for (int j=0; j < owners.length; j++) 
             msg.writeInt(owners[j]);
 
           msg.writeInt(dests.length);
           for (int j=0; j < dests.length; j++) 
             msg.writeInt(dests[j]);
 
           try {
             int mergingSize = mergingListeners.size();
             msg.writeInt(mergingSize);
             if (log.isDebugEnabled()) 
               log.debug("Number of states to get (mergingListeners): " + mergingSize);
             for (int j = 0; j < mergingSize; j++) {
               MergingListener mergingListener = (MergingListener) mergingListeners.get(j);
               msg.writeObject(mergingListener.getState(destIds));
             }
           } catch (Exception e) {
             log.warn("Exception caught when getting state from members", e);
           }
           msg.flush();
           multicastService.mcast(stream, null);
         } catch(Exception e) {
           log.warn("Exception during view change events of MergingData", e);
         }
       }
     } else {
       // Notify listeners of the new view if merging was not needed
       notifyView();
     } // END if (needupdate)
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipListener#hasLeft()
    */
   public void hasLeft()
   {
     for (MembershipListener listener : membershipListeners) {
       listener.hasLeft();
     }
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.MembershipListener#prepareChange()
    */
   public void prepareChange()
   {
     for (MembershipListener listener : membershipListeners) {
       listener.prepareChange();
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from MulticastListener
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /* (non-Javadoc)
    * @see jgroup.core.multicast.MulticastListener#getProtocolName()
    */
   public String getProtocolName() 
   {
     return PROTOCOL_NAME;
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.multicast.MulticastListener#deliverStream(java.io.InputStream, jgroup.core.MemberId, int)
    */
   public Object deliverStream(InputStream stream, MemberId sender, int seqNo)
   {
     if (log.isDebugEnabled()) {
       log.debug("MergingLayer: deliverStream from " + sender);
       log.debug(stream);
     }
 
     try {
       MarshalInputStream in = new MarshalInputStream(stream);
 
       MemberId[] owners = new MemberId[in.readInt()];
       for (int i = 0; i < owners.length; i++) {
         owners[i] = members[in.readInt()];
         if (log.isDebugEnabled()) {
           log.debug("owners[" + i + "]: " + owners[i] );
         }
         /*
          * If I belong to <owners>, the <mset> contains all the members
          * in the current view.
          */
         if (owners[i].equals(me)) {
           for (int k = 0; k < members.length; k++) {
             MemberInfo md = (MemberInfo) table.get(members[k]);
             md.mset = true;
           }
         }
 
         // Each member in <owners> are contained in <lset>.
         MemberInfo md = (MemberInfo) table.get(owners[i]);
         md.lset = true;
       }
 
       MemberId[] dests = new MemberId[in.readInt()];
       boolean updateState = false;
       for (int i = 0; i < dests.length; i++) {
         dests[i] = members[in.readInt()];
         if (dests[i].equals(me)) {
           // The state message is directed to me; perform putState()
           updateState = true;
         }
         if (log.isDebugEnabled()) {
           log.debug("dests[" + i + "]: " + dests[i] );
         }
       }
 
       if (!updateState) {
         if (log.isDebugEnabled())
           log.debug("Message not for me; ignoring the putState message");
         return null;
       }
       Object[] objs = new Object[in.readInt()];
       for (int j = 0; j < objs.length; j++) {
         objs[j] = in.readObject();
         if (log.isDebugEnabled())
           log.debug("Read STATE: " + objs[j]);
       }
 
       for (int j = 0, size = mergingListeners.size(); j < size; j++)
         ((MergingListener) mergingListeners.get(j)).putState(objs[j], dests);
     } catch (Exception e) {
       /*
        * If we for some reason received an incorrectly formated message;
        * we simply log the raised exception and ignore the message since
        * we cannot know its origin.  That is it may be spam.
        */
       log.warn("MergingLayer: Unable to unmarshal stream message", e);
     }
 
     remainingPutStates--;
     if (log.isDebugEnabled())
       log.debug("Remaining put state calls: " + remainingPutStates);
     if (remainingPutStates == 0) {
       // Notify listeners of the new view (only after updating the the server state)
       notifyView();
     }
     return null;
   }
 
 
   /* (non-Javadoc)
    * @see jgroup.core.multicast.MulticastListener#deliverObject(java.lang.Object)
    */
   public Object deliverObject(Object msg, MemberId sender, int seqNo)
   {
     throw new UnsupportedOperationException();
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Private Methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * 
    */
   private void notifyView()
   {
     MembershipLayer.notifyViewListeners(view, previousView, membershipListeners);
     previousView = view;
   }
 
   /**
    *
    */
   private int[] getOwners(MemberInfo[] mdata)
   {
     // Count owners
     int len = 0;
     for (int i=0; i < mdata.length; i++)
       if (mdata[i].lset) len++;
     
     // Write owners
     int[] owners = new int[len];
     len = 0;
     for (int i=0; i < mdata.length; i++) 
       if (mdata[i].lset) {
         owners[len] = table.getIndex(mdata[i].id);
         len++;
       }
 
     return owners;
   }
     
   /**
    * Retreive the set of members that needs to be updated with
    * state information from the local coordinator.
    */
   private int[] getDests(MemberInfo[] mdata)
   {
     // Count dests
     int len = 0;
     for (int i=0; i < mdata.length; i++)
       if (!mdata[i].mset && mdata[i].tset) len++;
 
     // Write dests
     int[] dests = new int[len];
     len = 0;
     for (int i=0; i < mdata.length; i++)
       if (!mdata[i].mset && mdata[i].tset) {
         dests[len] = table.getIndex(mdata[i].id);
         len++;
       }
 
     return dests;
   }
 
 
   /**
    * Returns true if the local member is elected as coordinator for
    * its partition.  Note that we assume that the input array is
    * ordered such that the first indecies contain members of the
    * most recent view.
    */
   private boolean hasLocalCoordinator(MemberInfo[] mdata)
   {
     /*
      * Search for the first member in mdata with mset equal to true.
      */
     int min = 0;
     for (int i=0; i < mdata.length; i++) {
       if (mdata[i].mset && table.getStatus(mdata[i].id) != MemberTable.CRASHED) {
         min = i;
         break;
       }
     }
     if (log.isDebugEnabled())
       log.debug("Elected Coordinator: " + mdata[min].id);
     return (mdata[min].id.equals(me));
   }
 
 
   private class MemberInfo 
   { 
     /* True if the member may potentially know LESS updates (up-to-date set) */
     boolean lset;
 
     /* True if the member may potentially know MORE updates */
     boolean mset;
 
     /* True if this member is in the most recent view (vcomp) */
     boolean tset;
 
     MemberId id;
 
     MemberInfo(MemberId id, boolean set)
     {
       this.id = id;
       lset = set;
       mset = set;
       tset = set;
     }
 
     @Override
     public String toString()
     {
       StringBuilder buf = new StringBuilder("MergeInfo: ");
       buf.append(table.toString(id));
       buf.append(": mset=");
       buf.append(mset);
       buf.append(", lset=");
       buf.append(lset);
       buf.append(", tset=");
       buf.append(tset);
       return buf.toString();
     }
 
   } // END MemberInfo
 
 } // END MergingLayer