/*
    * Copyright (c) 1998-2006 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.gmi.protocols;
  
  import java.io.Externalizable;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.ObjectInput;
  import java.io.ObjectOutput;
  import java.io.OutputStream;
  import java.rmi.RemoteException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.NoSuchElementException;
  import java.util.SortedMap;
  import java.util.TreeMap;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  import jgroup.core.MemberId;
  import jgroup.core.MembershipService;
  import jgroup.core.multicast.MulticastListener;
  import jgroup.core.multicast.MulticastService;
  import jgroup.relacs.gmi.GroupAckListener;
  import jgroup.relacs.gmi.GroupInvocationDispatcher;
  import jgroup.relacs.gmi.InvocationResult;
  import jgroup.relacs.gmi.MethodSemantics;
  import jgroup.relacs.types.MemberIdImpl;
  import jgroup.util.InMessage;
  import jgroup.util.OutMessage;
  
  import org.apache.log4j.Logger;
  import org.apache.log4j.MDC;
  
  /**
   * Atomic represents a protocol that dispatch method invocations
   * received through the local endpoint to all the other methods of
   * the group, including the local server.  Each message received is
   * ordered according to a total order among all members of the group.
   * The result from all members is returned to the local endpoint whom
   * first received the method invocation request from some client, and
   * one of results are selected and passed back to the client.
   * 
   * Total ordering is accomplished using distributed agreement protocol similar
   * to that of [Birman and Joseph 1987].
   *
   * @author Hein Meling <hein.meling@uis.no>
   * @since Jgroup 2.2
   */
  public class Atomic
    extends BasicDispatcher
    implements MulticastListener
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(Atomic.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Constants
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Protocol name for demux in the multicast layer */
    private static final String PROTOCOL_NAME = MethodSemantics.ATOMIC.toString();
  
    /** Message type identifiers for the distributed agreement protocol */
    private enum MsgType { REQUEST_SEQ_NO, AGREED_SEQ_NO };
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** The lock preventing concurrent event handling */
    private final Lock lock = new ReentrantLock();
 
   /** The multicast service interface */
   private final MulticastService mcast;
 
   /** The membership service interface */
   private final MembershipService pgms;
 
   /** This member's identifier */
   private final MemberId me;
 
   /** The largest agreed sequence number */
   private SequenceNo largestAgreedSeqNo;
 
   /** The largest proposed sequence number */
   private SequenceNo largestProposedSeqNo;
 
   /** Map of proposals: MessageID -> SequenceNo (Proposal) */
   private final Map<MessageID, SequenceNo> proposalsMap =
     new HashMap<MessageID, SequenceNo>();
 
   /** The hold-back queue of messages (mapping: SequenceNo -> InputStream) */
   private final SortedMap<SequenceNo, InputStream> holdBackQueue =
     new TreeMap<SequenceNo, InputStream>();
 
   /** Map of result collectors */
   private final Map<SequenceNo, ResultCollector> resultCollectors =
     Collections.synchronizedMap(new HashMap<SequenceNo, ResultCollector>());
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructor
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   public Atomic(GroupInvocationDispatcher dispatcher,
       MulticastService mcast, MembershipService pgms)
   {
     super(dispatcher);
     this.mcast = mcast;
     this.pgms = pgms;
     this.me = pgms.getMyIdentifier();
     largestAgreedSeqNo = new SequenceNo(me);
     largestProposedSeqNo = new SequenceNo(me);
     /*
      * We need to listen to multicast events, and since this is
      * not a layer it will not be handled automatically by the
      * group manager construction mechanism.
      */
     mcast.addListener(this);
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // ProtocolDispatcher interface (overriding BasicDispatcher)
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * Handle inbound request to the local endpoint with <i>atomic</i>
    * invocation semantics.
    * 
    * The local endpoint will send a multicast message to all the other group
    * members for method invocation dispatching at each members.  They each
    * return a result to the local endpoint, and one of these values are selected
    * and returned to the client (GroupInvocationHandler).
    * 
    * @see jgroup.relacs.gmi.protocols.ProtocolDispatcher#dispatch(java.io.InputStream)
    */
   public InvocationResult dispatch(InputStream in)
     throws IOException
   {
     // Reduce overhead if group has a single member only
     if (pgms.members() == 1)
       return super.dispatch(in);
 
     SequenceNo agreed = getAgreed(in);
     if (agreed == null)
       throw new IOException("Could not obtain agreed sequence number; Proposals not received from all members");
 
     OutputStream agreedMsg = prepareAgreedMessage(agreed);
     ResultCollector rc = new ResultCollector(agreed);
     resultCollectors.put(agreed, rc);
 
     // send out the agreed message (eventually causing delivery of method invocations)
     try {
       mcast.mcast(agreedMsg, rc.getRemoteAckListener());
     } catch (Exception e) {
       log.warn("Failed multicast agreed sequence numbers to the group", e);
       IOException ioex = new IOException("Failed multicast agreed sequence numbers to the group");
       ioex.initCause(e);
       throw ioex;
     }
     // Wait for ack listener
     return (InvocationResult) rc.getResult();
   }
 
 
   /**
    * 
    * @param agreed
    * @throws IOException
    */
   private OutputStream prepareAgreedMessage(SequenceNo agreed)
     throws IOException
   {
     /* Send the agreed sequence number to all members */
     OutMessage out = (OutMessage) mcast.getMessage(PROTOCOL_NAME);
     // write common part: message type
     out.writeObject(MsgType.AGREED_SEQ_NO);
     // write the agreed sequence number
     agreed.writeExternal(out);
     out.flush();
     return out;
   }
 
 
   /**
    * Multicast the invocation to the group, and wait for proposals for the
    * sequence number of this invocation; the greatest proposal is returned
    * as the agreed sequence number.
    * 
    * @param in the invocation stream to multicast to the group
    * @return the agreed sequence number (the greatest proposal)
    * @throws IOException if the invocation failed due to either a communication
    *   failure or a new view was installed preventing all members from sending
    *   their propsal.
    */
   private SequenceNo getAgreed(InputStream in)
     throws IOException
   {
     OutputStream reqMsg = prepareReqMessage(in);
     // Do multicast the invocation; wait for ordering proposals
     ProposalCollector pc = new ProposalCollector();
     if (log.isDebugEnabled())
       log.debug("Atomic: multicasting invocation to group members");
     try {
       mcast.mcast(reqMsg, pc.getRemoteAckListener());
     } catch (Exception e) {
       log.warn("Invocation failed", e);
       IOException ioex = new IOException("Failed to send invocation to group");
       ioex.initCause(e);
       throw ioex;
     }
     // Wait for all proposals and compute the agreed sequence number
     return pc.getGreatestProposal();
   }
 
 
   /**
    * 
    * @param in
    * @return
    * @throws IOException
    */
   private OutputStream prepareReqMessage(InputStream in)
     throws IOException
   {
     OutMessage mout = (OutMessage) mcast.getMessage(PROTOCOL_NAME);
     mout.writeObject(MsgType.REQUEST_SEQ_NO);
     /* Create a message id for this message and write it to the stream header */
     MessageID msgId = new MessageID(me);
     msgId.writeExternal(mout);
     // Write request into multicast output stream
     byte[] buf = new byte[1500];
     int bytesRead;
     while ((bytesRead = in.read(buf)) != -1) {
       mout.write(buf, 0, bytesRead);
     }
     return mout;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // MulticastListener methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /* (non-Javadoc)
    * @see jgroup.core.multicast.MulticastListener#getProtocolName()
    */
   public String getProtocolName()
   {
     return PROTOCOL_NAME;
   }
 
   /**
    * Received a multicast message from a mediating endpoint for invocation
    * dispatching at the local server.
    * 
    * @param msg
    *   Atomic multicast message received
    * @return
    *   The result of the atomic multicast invocation, or <code>null</code>
    *   if the invocation failed.
    * 
    * @see jgroup.core.multicast.MulticastListener#deliverStream(java.io.InputStream, jgroup.core.MemberId, int)
    */
   public Object deliverStream(InputStream msg, MemberId sender, int seqNo)
   {
     if (log.isDebugEnabled()) {
       MDC.put("group", "[Group: " + pgms.getGid() + "]");
       log.debug("deliverStream() from " + sender);
     }
     InMessage inmsg = (InMessage) msg;
     try {
       /* Read message type from input stream */
       MsgType msgType = (MsgType) inmsg.readObject();
       /* Multiplex on the various message types */
       switch (msgType) {
         case REQUEST_SEQ_NO:
           return computeProposedSeqNo(inmsg);
 
         case AGREED_SEQ_NO:
           return dispatchAgreedInvocations(inmsg);
 
         default:
           log.error("Unknown message type: " + msgType);
           break;
       }
     } catch (IOException e) {
       log.warn("Failed to read from input stream", e);
     } catch (ClassNotFoundException e) {
       log.warn("Class not found", e);
     }
 
     return null;
   }
 
 
   private Object dispatchAgreedInvocations(InMessage inmsg)
     throws IOException, ClassNotFoundException
   {
     /* Read the agreed sequence number from the input stream */
     SequenceNo agreed = new SequenceNo();
     agreed.readExternal(inmsg);
     if (log.isDebugEnabled())
       log.debug("Received AGREED: " + agreed + " for msg " + agreed.msgId);
 
     // Reorder the holdBackQueue if agreed differs from my proposed sequence number
     reorderQueue(agreed);
     if (log.isDebugEnabled()) {
       log.debug("holdBackQueue=" + holdBackQueue.keySet());
       log.debug(" proposalsMap=" + proposalsMap);
     }
 
     return dispatchFromQueue(agreed);
   }
 
 
   /**
    * 
    * @param agreed
    */
   private Object dispatchFromQueue(SequenceNo agreed)
   {
     List<SequenceNo> results = new ArrayList<SequenceNo>(holdBackQueue.size());
     lock.lock();
     try {
       for (Iterator<Map.Entry<SequenceNo, InputStream>> iter = holdBackQueue.entrySet().iterator(); iter.hasNext();) {
         Map.Entry<SequenceNo, InputStream> entry = iter.next();
         SequenceNo seqno = entry.getKey();
         // if this is a newly agreed seqno, tag the message as deliverable
         if (seqno.equals(agreed))
           seqno.markAsDeliverable();
         if (seqno.isDeliverable()) {
           InputStream stream = entry.getValue();
           iter.remove();
           if (log.isDebugEnabled())
             log.debug("Delivering message: " + seqno.msgId + ", assigned seqno: " + seqno);
           try {
             // Perform the actual method invocation on the local server (BasicDispatcher)
             seqno.result = super.dispatch(stream);
             /*
              * Save the sequence number in the results list (the result object returned
              * by the invocation is embedded within the sequence number object)
              */
             results.add(seqno);
           } catch (IOException e) {
             log.warn("Failed to dispatch atomic invocation", e);
           }
         } else {
           // Stop moving messages to the delivery queue
           if (log.isDebugEnabled())
             log.debug("Stopping message delivery. " + seqno.msgId + " is undeliverable: " + seqno);
           break;
         }
       }
     } finally {
       lock.unlock();
     }
     if (log.isDebugEnabled())
       log.debug("results.size=" + results.size());
     return results;
   }
 
 
   /**
    * Reorder the hold-back queue in case the given agreed sequence number is
    * different from the sequence number proposed by this local member.
    * 
    * @param agreed the sequence number that all have agreed to use for its associated message
    */
   private void reorderQueue(SequenceNo agreed)
   {
     lock.lock();
     try {
       // Update the largest agreed sequence number
       largestAgreedSeqNo = SequenceNo.max(largestAgreedSeqNo, agreed);
       // Get my proposal for the msgId currently being ordered
       SequenceNo myProposal = proposalsMap.remove(agreed.msgId);
       if (!agreed.equals(myProposal)) {
         // Reorder the hold-back queue
         if (log.isDebugEnabled())
           log.debug("Reordering hold-back queue: myProposal=" + myProposal
               + " is different from agreed=" + agreed);
         InputStream stream = holdBackQueue.remove(myProposal);
         if (stream == null) {
           log.error("MESSAGE NOT FOUND IN HOLD-BACK QUEUE: " + myProposal);
           // abort message handling
           throw new NoSuchElementException("Message " + myProposal + " not found in hold-back queue");
         } else {
           holdBackQueue.put(agreed, stream);
         }
       }
     } finally {
       lock.unlock();
     }
   }
 
 
   /**
    * 
    * @param inmsg
    * @return
    * @throws IOException
    * @throws ClassNotFoundException
    */
   private SequenceNo computeProposedSeqNo(InMessage inmsg)
     throws IOException, ClassNotFoundException
   {
     /* Read the message identifier from the input stream */
     MessageID msgId = new MessageID();
     msgId.readExternal(inmsg);
     lock.lock();
     try {
       /* Compute a new proposed sequence number */
       largestProposedSeqNo = SequenceNo.max(largestAgreedSeqNo, largestProposedSeqNo);
       largestProposedSeqNo.updateSeqNo(msgId);
       /* Associate the msg id and its proposed sequence number */
       proposalsMap.put(msgId, largestProposedSeqNo);
       /*
        * When getting the message back from the holdBackQueue, the method
        * associated with it can be dispatched in the delivery phase.
        */
       holdBackQueue.put(largestProposedSeqNo, inmsg);
       if (log.isDebugEnabled())
         log.debug("Computed PROPOSAL: " + largestProposedSeqNo + " for msg " + msgId);
       return largestProposedSeqNo;
     } finally {
       lock.unlock();
     }
 
   }
 
 
   /* (non-Javadoc)
    * @see jgroup.core.multicast.MulticastListener#deliverObject(java.lang.Object, jgroup.core.MemberId, int)
    */
   public Object deliverObject(Object msg, MemberId sender, int seqNo)
   {
     throw new UnsupportedOperationException();
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Nested classes extending the GroupAckListener
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * An <code>AckListener</code> for collecting proposals for the
    * distributed agreement protocol.
    */
   private class ProposalCollector
     extends GroupAckListener
   {
 
     public ProposalCollector()
     {
       super(new Object());
     }
 
     private boolean isCompleted(int index)
     {
       return (completed[index] && !(results[index] instanceof RemoteException));
     }
 
     /**
      * Returns the greateset proposal obtained from the members.
      * This method might return <code>null</code> in case not all
      * members responded with a <i>proposal</i>.
      */
     public SequenceNo getGreatestProposal()
     {
       pendingCompletion();
 
       // Select the greatest proposal
       SequenceNo greatestProposal = null;
       if (isCompleted(0)) {
         greatestProposal = (SequenceNo) results[0];
         for (int i = 1; i < results.length; i++) {
           if (isCompleted(i)) {
             SequenceNo theProposal = (SequenceNo) results[i];
             greatestProposal = SequenceNo.max(theProposal, greatestProposal);
           } else {
             greatestProposal = null;
             break;
           }
         }
       }
       if (greatestProposal == null)
         log.warn("Not all members have provided a PROPOSAL");
       return greatestProposal;
     }
 
   } // END ProposalCollector
 
 
   /**
    * An <code>AckListener</code> for collecting results for the
    * distributed agreement protocol and delivering to the application.
    */
   private class ResultCollector
     extends GroupAckListener
   {
 
     private SequenceNo agreed;
 
     public ResultCollector(SequenceNo agreed)
     {
       super(agreed);
       this.agreed = agreed;
     }
 
     /* (non-javadoc)
      * @see jgroup.core.multicast.AckListener#ack(MemberId, int, Object)
      */
     @SuppressWarnings("unchecked")
     @Override
     public void ack(MemberId id, int pos, Object obj)
       throws RemoteException
     {
       if (log.isDebugEnabled())
         log.debug("Recieved result from " + id + ", pos=" + pos + ": " + obj);
       if (log.isDebugEnabled())
         log.debug("resultCollectors: " + resultCollectors.keySet());
 
       List<SequenceNo> results = (List) obj;
       if (results.isEmpty()) {
         /*
          * Nothing was dispatched, and hence no result to deliver; may have
          * been delivered previously when two or more results was included?!?
          */
         if (log.isDebugEnabled())
           log.debug("Empty results list when agreed=" + agreed);
         ResultCollector rc = resultCollectors.get(agreed);
         // ack with null to ensure progress
         rc.doAck(id, pos, null);
         return;
       }
       for (SequenceNo seqno : results) {
         ResultCollector rc = resultCollectors.get(seqno);
         if (log.isDebugEnabled())
           log.debug("Delivering seqno=" + seqno + ", when agreed=" + agreed);
         rc.doAck(id, pos, seqno.result);
       }
     }
 
     private void doAck(MemberId id, int pos, Object obj)
       throws RemoteException
     {
       super.ack(id, pos, obj);
     }
 
     @Override
     public Object getResult()
     {
       // blocks until invocation completed
       Object result = super.getResult();
       // remove the result collector for this agreed sequence number (it has completed)
       resultCollectors.remove(agreed);
       return result;
     }
 
   } // END ResultCollector
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Nested static class for total order sequence numbers
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * Sequence number class used to define a total order of messages.
    */
   private static final class SequenceNo
     implements Externalizable, Comparable<SequenceNo>
   {
 
     private static final long serialVersionUID = -6394934923836273234L;
 
     /** The sequence number value */
     private int sequenceNo = 100;
 
     /** The member that generated this sequence number */
     private MemberId member;
 
     /**
      * The message identifier associated with this sequence number.
      * Note that the message identifier is not used when comparing
      * sequence numbers or when computing the hash code.
      */
     private MessageID msgId = null;
 
     /**
      * Result object associated with this sequence number.
      */
     private Object result = null;
 
     /** Set to true when the message can be delivered */
     private transient boolean deliverable = false;
 
 
     /** Default constructor for externalization */
     public SequenceNo() {}
 
     public SequenceNo(MemberId me)
     {
       member = me;
     }
 
     public SequenceNo(SequenceNo oldSeq)
     {
       this.sequenceNo = oldSeq.sequenceNo;
       this.member = oldSeq.member;
       this.msgId = oldSeq.msgId;
     }
 
     public static SequenceNo max(SequenceNo s1, SequenceNo s2)
     {
       if (s1.compareTo(s2) > 0) {
         return new SequenceNo(s1);
       } else {
         return new SequenceNo(s2);
       }
     }
 
     /**
      * Update the sequence number counter and set the message identifier associated
      * with this sequence number.
      * 
      * @param msgId the message identifier to which this sequence number should be associated.
      */
     public void updateSeqNo(MessageID msgId)
     {
       sequenceNo++;
       this.msgId = msgId;
     }
 
     public boolean isDeliverable()
     {
       return deliverable;
     }
 
     public void markAsDeliverable()
     {
       deliverable = true;
     }
 
     public int hashCode()
     {
       return sequenceNo ^ member.hashCode();
     }
 
     public boolean equals(Object obj)
     {
       if (obj instanceof SequenceNo) {
         SequenceNo sno = (SequenceNo) obj;
         return (sequenceNo == sno.sequenceNo && member.equals(sno.member));
       } else {
         return false;
       }
     }
 
     public String toString()
     {
       StringBuilder buf = new StringBuilder();
       buf.append("[");
       String a = member.getEndPoint().getAddress().getHostAddress().toString();
       String[] b = a.split("\\.");
       if (b.length > 0)
         buf.append(b[b.length-1]);
       buf.append(".");
       buf.append(sequenceNo);
       buf.append("]");
       return buf.toString();
     }
 
     /* (non-Javadoc)
      * @see java.io.Externalizable#readExternal(java.io.ObjectInput)
      */
     public void readExternal(ObjectInput in)
       throws IOException, ClassNotFoundException
     {
       sequenceNo = in.readInt();
       member = new MemberIdImpl();
       member.readExternal(in);
       msgId = new MessageID();
       msgId.readExternal(in);
       result = in.readObject();
     }
 
     /* (non-Javadoc)
      * @see java.io.Externalizable#writeExternal(java.io.ObjectOutput)
      */
     public void writeExternal(ObjectOutput out)
       throws IOException
     {
       out.writeInt(sequenceNo);
       member.writeExternal(out);
       msgId.writeExternal(out);
       out.writeObject(result);
     }
 
     /**
      * Compare two <code>SequenceNo</code> instances.
      *
      * For equal sequence number values we use the <code>MemberId</code>
      * to distinguish them.  Otherwise, we use only the sequence
      * number.  That is, the sequence number has precedence when
      * comparing two <code>SequenceNo</code> instances.
      *
      * @see java.lang.Comparable#compareTo(java.lang.Object)
      */
     public int compareTo(SequenceNo seqno)
     {
       if (this == seqno)
         return 0;
       if (this.sequenceNo == seqno.sequenceNo)
         return this.member.compareTo(seqno.member);
       else if (this.sequenceNo > seqno.sequenceNo)
         return 1;
       else
         return -1;
     }
 
   } // END SequenceNo
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Nested static class for message identifiers
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * A message identifier is a unique identifier for messages
    * generated within the system.  This is implement by a locally
    * unique sequence number and a member identifier.
    */
   private static final class MessageID
     implements Externalizable
   {
 
     private static final long serialVersionUID = -6073546790683304835L;
 
     private static long nextSequenceNo = 0;
     private long sequenceNo;
     private MemberId memberId;
 
     /** Default constructor for externalization */
     public MessageID() {}
 
     public MessageID(MemberId id)
     {
       this.sequenceNo = nextSequenceNo++;
       this.memberId = id;
     }
 
     public boolean equals(Object obj)
     {
       if (obj instanceof MessageID) {
         MessageID msgId = (MessageID) obj;
         return (sequenceNo == msgId.sequenceNo && memberId.equals(msgId.memberId));
       } else {
         return false;
       }
     }
 
     public int hashCode()
     {
       return (int)(sequenceNo ^ (sequenceNo >>> 32)) ^ memberId.hashCode();
     }
 
     public String toString()
     {
       StringBuilder buf = new StringBuilder();
       buf.append("[");
       String a = memberId.getEndPoint().getAddress().getHostAddress().toString();
       String[] b = a.split("\\.");
       if (b.length > 0)
         buf.append(b[b.length-1]);
       buf.append("*");
       buf.append(sequenceNo);
       buf.append("]");
       return buf.toString();
     }
 
     /* (non-Javadoc)
      * @see java.io.Externalizable#readExternal(java.io.ObjectInput)
      */
     public void readExternal(ObjectInput in)
       throws IOException, ClassNotFoundException
     {
       sequenceNo = in.readLong();
       memberId = new MemberIdImpl();
       memberId.readExternal(in);
     }
 
     /* (non-Javadoc)
      * @see java.io.Externalizable#writeExternal(java.io.ObjectOutput)
      */
     public void writeExternal(ObjectOutput out)
       throws IOException
     {
       out.writeLong(sequenceNo);
       memberId.writeExternal(out);
     }
 
   } // END MessageID
 
 } // END Atomic