/*
    * 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.daemon;
  
  import java.rmi.NoSuchObjectException;
  import java.rmi.RemoteException;
  import java.util.Collections;
  import java.util.LinkedList;
  import java.util.Queue;
  
  import jgroup.core.MemberId;
  import jgroup.core.View;
  import jgroup.core.multicast.AckListener;
  
  import org.apache.log4j.Logger;
  
  /**
   *  Buffer for managing multicast (<code>MsgMcast</code>) messages, and
   *  corresponding reply messages (<code>MsgResult</code>).  That
   *  includes also acknowledgement management.
   *
   *  @author Alberto Montresor
   *  @since Jgroup 0.8
   */
  final class SCopyOfSendBuffer
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(SCopyOfSendBuffer.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    @SuppressWarnings("unchecked")
    private final Queue<MsgMcast> buffer = (Queue) Collections.synchronizedList(new LinkedList<MsgMcast>());;
    @SuppressWarnings("unchecked")
    private final Queue<MsgResult> ackbuffer = (Queue) Collections.synchronizedList(new LinkedList<MsgResult>());
    private View        view;
    private long        vid;
    private int         hlen;
    private int         nmessages;
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Constructor 
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    SCopyOfSendBuffer() {}
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Methods
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /**
     *
     */
    void setView(View view, int hlen)
    {
      if (log.isDebugEnabled())
        log.debug("Flushing buffers");
      this.view = view;
      this.hlen = hlen;
      vid = view.getVid();
  
      nmessages = 0;
      
      // Call nack
      for (MsgMcast msg : buffer) {
        AckListener ackl = msg.ackl;
        if (ackl != null) { 
          try {
            if (log.isDebugEnabled())
              log.debug("viewChange to AckListener");
            ackl.viewChange();
            if (log.isDebugEnabled())
              log.debug("viewChange to AckListener completed");
          } catch (NoSuchObjectException e) {
           /*
            * Ignored, as this can happen when the AckListener has been
            * unexported since all results has been received correctly,
            * and thus the receiver does not need to consider a view change
            * for the current multicast message (invocation).
            */
         } catch (RemoteException e) {
           log.warn("Failed to notify the AckListener of view change: " + ackl, e);
         }
       }
     }
     buffer.clear();
     MsgResult msgr;
     while ((msgr = (MsgResult) ackbuffer.poll()) != null) {
       if (msgr.vid == vid) 
         insertMsgResult(msgr);
     }
   }
   
   
   /**
    *
    */
   void insertMsgMcast(MsgMcast msg)
   {
     try {
       if (msg.ackl != null) {
         if (log.isDebugEnabled())
           log.debug("notifyView to AckListener");
         msg.ackl.notifyView(view);
         if (log.isDebugEnabled())
           log.debug("notifyView completed");
       }
     } catch (RemoteException e) {
       log.warn("Failed to notify the AckListener of view change.", e);
     }
     buffer.add(msg);
     msg.setAckArray(hlen);
   }
 
   
   /**
    *
    */
   void insertMsgResult(MsgResult result)
   {
     if (result.vid == vid) {
       if (log.isDebugEnabled())
         log.debug("Inserting result (same view): " + result);
       for (MsgMcast message : buffer) {
         if (message.mid <= result.dlvr)
           message.setAck(result.hpos);
         if (message.mid == result.mid && message.sender.equals(result.sender)) {
           MemberId[] members = view.getMembers();
           log.assertLog(members.length > result.vpos,
               "BAD: Trying to send ack, but view seems to have less members: " + view);
           try {
             if (message.ackl != null)
               message.ackl.ack(members[result.vpos], result.vpos, result.result);
           } catch (RemoteException e) {
             log.warn("Failed to ACK the message.", e);
           }
         }
       }
       removeAcknowledged();
     } else {
       if (log.isDebugEnabled())
         log.debug("Inserting result (different view): " + result);
       ackbuffer.add(result);
     }
   }
 
 
   /**
    *  Marks the given host as suspected.  This will mark messages in
    *  this buffer, that was sent to the given host, as acknowledged.
    *
    *  @param host
    *    The host to be marked as suspected.
    */
   void suspect(HostData host)
   {
     if (host.hasValidViewIndex()) {
       int hpos = host.getViewIndex();
       for (MsgMcast message : buffer) {
         message.setAck(hpos);
       }
       removeAcknowledged();
     }
   }
 
 
   /**
    *  Remove messages that have been already acknowledged by all other
    *  host.
    */
   private void removeAcknowledged()
   {
     MsgMcast    message;
     while ((message = (MsgMcast) buffer.peek()) != null) {
       if (message.isComplete()) 
         buffer.poll();
       else
         break;
     }
   }
 
 
   int getMid()
   {
     return ++nmessages;
   }
 
   boolean isEmpty()
   {
     return buffer.isEmpty();
   }
   
   
 } // END SendBuffer