/*
    * Copyright  1999-2004 The Apache Software Foundation.
    *
    *  Licensed under the Apache License, Version 2.0 (the "License");
    *  you may not use this file except in compliance with the License.
    *  You may obtain a copy of the License at
    *
    *      http://www.apache.org/licenses/LICENSE-2.0
    *
   *  Unless required by applicable law or agreed to in writing, software
   *  distributed under the License is distributed on an "AS IS" BASIS,
   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *  See the License for the specific language governing permissions and
   *  limitations under the License.
   *
   */
  
  package org.apache.sandesha.storage.queue;
  
  import org.apache.axis.components.logger.LogFactory;
  import org.apache.commons.logging.Log;
  import org.apache.sandesha.RMMessageContext;
  import org.apache.sandesha.util.PolicyLoader;
  
  import java.util.*;
  
  /*
   * Created on Aug 4, 2004 at 5:08:29 PM
   */
  
  /***
   * @author Chamikara Jayalath
   * @author Jaliya Ekanayaka
   */
  
  /***
   * This class works as a hash map for storing response messages until they are
   * sent.
   */
  public class OutgoingSequence extends AbstractSequence {
  
      private String outSequenceId;
      private boolean outSeqApproved;
      private HashMap hash;
      private ArrayList markedAsDelete;
      private ArrayList sendMsgNoList;
      private long lastMsgNo = -1;
      private long nextAutoNumber; // key for storing messages.
      private static final Log log = LogFactory.getLog(OutgoingSequence.class.getName());
      public boolean terminateSent = false;
      private boolean hasResponse = false;
  
      public boolean hasResponse() {
          return hasResponse;
      }
  
      public void setHasResponse(boolean hasResponse) {
          this.hasResponse = hasResponse;
      }
  
      public boolean isTerminateSent() {
          return terminateSent;
      }
  
      public void setTerminateSent(boolean terminateSent) {
          this.terminateSent = terminateSent;
      }
  
      public OutgoingSequence(String sequenceId) {
          this.sequenceId = sequenceId;
          hash = new HashMap();
          markedAsDelete = new ArrayList();
          nextAutoNumber = 1; //This is the key for storing messages.
          outSeqApproved = false;
          sendMsgNoList = new ArrayList();
      }
  
      /*
       * public boolean hasMessagesToSend(){ return hasMessagesToSend; }
       */
  
  
      public boolean isOutSeqApproved() {
          return outSeqApproved;
      }
  
      public void setOutSeqApproved(boolean b) {
          outSeqApproved = b;
      }
  
      public String getOutSequenceId() {
          return outSequenceId;
      }
  
      public void setOutSequenceId(String string) {
          outSequenceId = string;
      }
  
      /***
      * adds the message to map.
      */
     public Object putNewMessage(RMMessageContext msg) {
         Long key = new Long(nextAutoNumber);
         Object obj = hash.put(key, msg);
         increaseAutoNo();
         return obj;
     }
 
     /***
      * Increases auto number by 1.
      */
     private void increaseAutoNo() {
         nextAutoNumber++;
     }
 
     /***
      * Returns the next deliverable message if has any. Otherwise returns null.
      */
     public RMMessageContext getNextMessageToSend() {
         RMMessageContext minMsg = null;
         Iterator keys = hash.keySet().iterator();
 
         whileLoop: while (keys.hasNext()) {
             RMMessageContext tempMsg;
             tempMsg = (RMMessageContext) hash.get(keys.next());
             Long msgNo = new Long(tempMsg.getMsgNumber());
             if (markedAsDelete.contains(msgNo)) {
                 continue;
             }
             long lastSentTime = tempMsg.getLastSentTime();
             Date d = new Date();
             long currentTime = d.getTime();
 
             long retransmissionInterval = PolicyLoader.getInstance().getBaseRetransmissionInterval();
             if (currentTime >= lastSentTime + retransmissionInterval) {
                 if (minMsg == null)
                     minMsg = tempMsg;
                 else {
                     long msgNo1, msgNo2;
                     msgNo1 = tempMsg.getMsgNumber();
                     msgNo2 = minMsg.getMsgNumber();
                     if (msgNo1 < msgNo2)
                         minMsg = tempMsg;
                 }
             }
         }
 
         Date d = new Date();
         long time = d.getTime();
         if (minMsg != null) {
             minMsg.setLastSentTime(time);
         }
 
         return minMsg;
     }
 
     public boolean hasMessage(Long key) {
         Object obj = hash.get(key);
 
         return (!(obj == null));
     }
 
     public void clearSequence(boolean yes) {
         if (!yes)
             return;
         hash.clear();
         nextAutoNumber = 1;
         outSeqApproved = false;
         outSequenceId = null;
         sequenceId = null;
     }
 
     public Set getAllKeys() {
         return hash.keySet();
     }
 
     public String getMessageId(Long key) {
         RMMessageContext msg = (RMMessageContext) hash.get(key);
         if (msg == null)
             return null;
 
         return msg.getMessageID();
 
     }
 
     //Deleting returns the deleted message.
     public RMMessageContext deleteMessage(Long msgId) {
         RMMessageContext msg = (RMMessageContext) hash.get(msgId);
         if (msg == null)
             return null;
         hash.remove(msgId);
         return msg;
     }
 
     public boolean markMessageDeleted(Long messageNo) {
         if (hash.containsKey(messageNo)) {
             markedAsDelete.add(messageNo);
             return true;
         }
         return false;
     }
 
     public long nextMessageNumber() {
         return nextAutoNumber;
     }
 
     public boolean isMessagePresent(String msgId) {
         boolean b = false;
         b = hash.containsKey(msgId);
         return b;
     }
 
     public boolean hasMessageWithId(String msgId) {
         Iterator it = hash.keySet().iterator();
         boolean result = false;
         while (it.hasNext()) {
             RMMessageContext msg = (RMMessageContext) hash.get(it.next());
             if (msg.getMessageID().equals(msgId)) {
                 result = true;
                 break;
             }
         }
         return result;
     }
 
     public List getReceivedMsgNumbers() {
         List result = new ArrayList();
         Iterator it = hash.keySet().iterator();
 
         while (it.hasNext()) {
             Object key = it.next();
             RMMessageContext msg = (RMMessageContext) hash.get(key);
             long l = msg.getMsgNumber();
             result.add(new Long(l));
         }
         return result;
     }
 
     public void setAckReceived(long msgNo) {
         RMMessageContext msg = (RMMessageContext) hash.get(new Long(msgNo));
         if (msg != null) {
             msg.setAckReceived(true);
         } else {
             log.error("ERROR: MESSAGE IS NULL IN ResponseSeqHash");
         }
 
     }
 
     public boolean isAckComplete() {
         long lastMsgNo = getLastMsgNumber();
         if (lastMsgNo <= 0) {
             return false;
         }
         Iterator it = hash.keySet().iterator();
         for (long i = 1; i < lastMsgNo; i++) {
             if (!hasMessage(new Long(i))) {
                 return false;
             }
         }
 
         it = hash.keySet().iterator();
         while (it.hasNext()) {
             RMMessageContext msg = (RMMessageContext) hash.get(it.next());
             if (!msg.isAckReceived()) {
                 return false;
             }
         }
         return true;
 
     }
 
     public void addMsgToSendList(long msgNo) {
         sendMsgNoList.add(new Long(msgNo));
     }
 
     public boolean isMsgInSentList(long msgNo) {
         return sendMsgNoList.contains(new Long(msgNo));
     }
 
     public boolean hasLastMsgReceived() {
         if (lastMsgNo > 0)
             return true;
 
         return false;
     }
 
     public long getLastMsgNumber() {
         if (lastMsgNo > 0)
             return lastMsgNo;
 
         return -1;
     }
 
     public void setLastMsg(long lastMsg) {
         lastMsgNo = lastMsg;
     }
 
 }