/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2025 ScalAgent Distributed Technologies
 * Copyright (C) 1996 - 2000 Dyade
 *
 * This library 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 any later version.
 * 
 * This library 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 library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): Frederic Maistre (INRIA)
 * Contributor(s): ScalAgent Distributed Technologies
 */
package org.objectweb.joram.mom.dest;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.Vector;

import javax.management.openmbean.CompositeData;
import javax.management.openmbean.TabularData;

import org.objectweb.joram.mom.messages.Message;
import org.objectweb.joram.mom.messages.MessageJMXWrapper;
import org.objectweb.joram.mom.messages.MessageView;
import org.objectweb.joram.mom.notifications.AbortReceiveRequest;
import org.objectweb.joram.mom.notifications.AbstractRequestNot;
import org.objectweb.joram.mom.notifications.AcknowledgeRequest;
import org.objectweb.joram.mom.notifications.BrowseReply;
import org.objectweb.joram.mom.notifications.BrowseRequest;
import org.objectweb.joram.mom.notifications.ClientMessages;
import org.objectweb.joram.mom.notifications.DenyRequest;
import org.objectweb.joram.mom.notifications.ExceptionReply;
import org.objectweb.joram.mom.notifications.FwdAdminRequestNot;
import org.objectweb.joram.mom.notifications.QueueDeliveryTimeNot;
import org.objectweb.joram.mom.notifications.QueueMsgReply;
import org.objectweb.joram.mom.notifications.ReceiveRequest;
import org.objectweb.joram.mom.notifications.TopicMsgsReply;
import org.objectweb.joram.mom.notifications.WakeUpNot;
import org.objectweb.joram.mom.util.DMQManager;
import org.objectweb.joram.mom.util.JoramHelper;
import org.objectweb.joram.mom.util.QueueDeliveryTimeTask;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.MessageErrorConstants;
import org.objectweb.joram.shared.admin.AdminReply;
import org.objectweb.joram.shared.admin.AdminRequest;
import org.objectweb.joram.shared.admin.ClearQueue;
import org.objectweb.joram.shared.admin.DeleteQueueMessage;
import org.objectweb.joram.shared.admin.GetDMQSettingsReply;
import org.objectweb.joram.shared.admin.GetDMQSettingsRequest;
import org.objectweb.joram.shared.admin.GetDeliveredMessages;
import org.objectweb.joram.shared.admin.GetNbMaxMsgRequest;
import org.objectweb.joram.shared.admin.GetNumberReply;
import org.objectweb.joram.shared.admin.GetPendingMessages;
import org.objectweb.joram.shared.admin.GetPendingRequests;
import org.objectweb.joram.shared.admin.GetQueueMessage;
import org.objectweb.joram.shared.admin.GetQueueMessageIds;
import org.objectweb.joram.shared.admin.GetQueueMessageIdsRep;
import org.objectweb.joram.shared.admin.GetQueueMessageRep;
import org.objectweb.joram.shared.admin.SetNbMaxMsgRequest;
import org.objectweb.joram.shared.admin.SetSyncExceptionOnFullDestRequest;
import org.objectweb.joram.shared.admin.SetThresholdRequest;
import org.objectweb.joram.shared.excepts.AccessException;
import org.objectweb.joram.shared.excepts.DestinationException;
import org.objectweb.joram.shared.excepts.MomException;
import org.objectweb.joram.shared.messages.ConversionHelper;
import org.objectweb.joram.shared.selectors.Selector;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Level;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.AdminSyncNotification;
import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.agent.Channel;
import fr.dyade.aaa.agent.DeleteNot;
import fr.dyade.aaa.agent.ExpiredNot;
import fr.dyade.aaa.agent.Notification;
import fr.dyade.aaa.agent.UnknownAgent;
import fr.dyade.aaa.common.Debug;
import fr.dyade.aaa.common.encoding.Decoder;
import fr.dyade.aaa.common.encoding.Encodable;
import fr.dyade.aaa.common.encoding.EncodableFactory;
import fr.dyade.aaa.common.encoding.Encoder;

/**
 * The <code>Queue</code> class implements the MOM queue behavior,
 * basically storing messages and delivering them upon clients requests.
 */
public class Queue extends Destination implements QueueMBean {
  /** define serialVersionUID for interoperability */
  private static final long serialVersionUID = 1L;

  static final Logger logger = Debug.getLogger(Queue.class.getName());
  static final Logger logmsg = Debug.getLogger(Queue.class.getName() + ".Message");
  
  public static final String DELIVERY_TABLE_PREFIX = "DT_";
  public static final String ARRIVAL_STATE_PREFIX = "AS_";

  /** Static value holding the default DMQ identifier for a server. */
  static AgentId defaultDMQId = null;

  /**
   * Property allowing to delay the denied messages even if there are not considered 'redelivered' (see JMS specification).
   * Normally, a message is considered 'redelivered' only if it has been delivered and rejected by the client. So, Messages
   * 'denied' due to an error in the delivery mechanism are immediately re-delivered.
   * By default false.
   */
  public final static String LENIENT_REDELIVERY_DELAY = "org.objectweb.joram.mom.messages.lenientRedeliveryDelay";
  private boolean lenientRedeliveryDelay = AgentServer.getBoolean(LENIENT_REDELIVERY_DELAY);
  
  /**
   * Property allowing to handle 'denied' messages as if they were considered 'redelivered' (see JMS specification).
   * Normally, a message is considered 'redelivered' only if it has been delivered and rejected by the client, in this case
   * its delivery counter is incremented. Then, a message is  handled as 'undeliverable' only if it has been delivered and
   * rejected several times by the client (the maximum number of attempts is defined by the queue's 'threshold' property).
   * This property allows to increment the message delivery counter even when the message has been 'denied' due to an error
   * in the delivery mechanism.
   * The default is false.
   */
  public final static String LENIENT_UNDELIVERABLE = "org.objectweb.joram.mom.messages.lenientUndeliverable";
  private boolean lenientUndeliverable = AgentServer.getBoolean(LENIENT_UNDELIVERABLE);
  
  /**
   * Threshold above which messages are considered as undeliverable because
   * constantly denied; 0 stands for no threshold, -1 for value not set.
   */
  private int threshold = -1;

  /** Static value holding the default threshold for a server. */
  static int defaultThreshold = -1;

  /**
   * Returns  the threshold value of this queue, -1 if not set.
   *
   * @return the threshold value of this queue; -1 if not set.
   */
  @Override
  public int getThreshold() {
    return threshold;
  }

  /**
   * Sets or unsets the threshold for this queue.
   *
   * @param threshold The threshold value to be set (-1 for unsetting previous value).
   */
  @Override
  public void setThreshold(int threshold) {
    this.threshold = threshold;
  }

  /** Static method returning the default threshold. */
  public static int getDefaultThreshold() {
    return defaultThreshold;
  }

  /** Static method returning the default DMQ identifier. */
  public static AgentId getDefaultDMQId() {
    return defaultDMQId;
  }

  /** 
   * The re-delivery delay in seconds use to wait before re-delivering 
   * messages after a deny. 
   */
  private int redeliveryDelay = 0;

  /** Static value holding the default redelivery delay for a server. */
  static int defaultRedeliveryDelay = -1;

  /**
   * Returns the delay in seconds use to wait before re-delivering messages
   * after a deny. 
   * 
   * @return the reDeliveryDelay
   */
  @Override
  public final int getRedeliveryDelay() {
    if (redeliveryDelay == 0)
      return getDefaultRedeliveryDelay();
    return redeliveryDelay;
  }

  /**
   * Sets the delay in seconds use to wait before re-delivering messages
   * after a deny.
   * 
   * @param redeliveryDelay the reDeliveryDelay to set
   */
  @Override
  public final void setRedeliveryDelay(int redeliveryDelay) {
    this.redeliveryDelay = redeliveryDelay;
    setSave();
  }

  /** Static method returning the default redelivery delay for a server. */
  public static final int getDefaultRedeliveryDelay() {
    return defaultRedeliveryDelay;
  }
  
  public static final void setDefaultRedeliveryDelay(int reDeliveryDelay) {
    defaultRedeliveryDelay = reDeliveryDelay;
  }
  
  /** 
   * The delivery delay in milliseconds used to wait before delivering a message.
   * If set the resulting delay is the max between this value and the message
   * property.
   */
  private int deliveryDelay = 0;

  /**
   * Returns the Queue deliveryDelay in milliseconds.
   * @return the DeliveryDelay
   */
  @Override
  public final int getDeliveryDelay() {
    return deliveryDelay;
  }

  /**
   * Sets the Queue deliveryDelay in milliseconds.
   * @param deliveryDelay the deliveryDelay to set
   */
  @Override
  public final void setDeliveryDelay(int deliveryDelay) {
    this.deliveryDelay = deliveryDelay;
    setSave();
  }

  private boolean pause = false;
  
  @Override
  public boolean isPause() {
    return pause;
  }
  
  @Override
  public void setPause(boolean pause) {
    if (this.pause == pause) return;
    
    this.pause = pause;
    setSave();
    // Sends a QueueDeliveryTimeNot in order to trigger message delivery.
    if (!pause)
      Channel.sendTo(getId(), new QueueDeliveryTimeNot(null, false));
  }
  
  /**
   * <code>true</code> if all the stored messages have the same priority.
   * Note: <code>messages</code> list is ordered by priorities, so we could test if
   * first and last message have the same priority.
   */
  private boolean samePriorities;
  
  /** Number of stored messages with an expiration date. */
  protected int nbExpirations;

  /** Common priority value. */
  private int priority; 

  /** Table keeping the message deliveries */
  protected transient QueueDeliveryTable deliveryTable;

  /** Counter of messages arrivals. */
  protected transient QueueArrivalState arrivalState;

  /** List holding the requests before reply or expiry. */
  protected List<ReceiveRequest> requests = new Vector<ReceiveRequest>();
  
  /**
   * Creates a queue.
   */
  public Queue() {
    super();
  }
  
  /**
   * Creates a queue with a specified stamp.
   * It is used by the Encodable framework as the default constructor allocates a new stamp.
   * 
   * @param name  Name of topic;
   * @param fixed If true, topic is fixed in memory.
   * @param stamp Specific stamp for resulting agent.
   */
  protected Queue(String name, boolean fixed, int stamp) {
    super(name, fixed, stamp);
  }
  
  /**
   * Configures an {@link Queue} instance.
   * 
   * @param properties
   *          The initial set of properties.
   */
  @Override
  public void setProperties(Properties properties, boolean firstTime) throws Exception {
    super.setProperties(properties, firstTime);
    
    // Set redeliveryDelay if defined.
    if (properties != null && properties.containsKey(DestinationConstants.REDELIVERY_DELAY)) {
      setRedeliveryDelay(ConversionHelper.toInt(properties.get(DestinationConstants.REDELIVERY_DELAY)));
    }
    // Set deliveryDelay if defined.
    if (properties != null && properties.containsKey(DestinationConstants.DELIVERY_DELAY)) {
      setDeliveryDelay(ConversionHelper.toInt(properties.get(DestinationConstants.DELIVERY_DELAY)));
    }
  }
  
  /**
   * Distributes the received notifications to the appropriate reactions.
   * 
   * @throws Exception
   */
  public void react(AgentId from, Notification not) throws Exception {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.react(" + from + ',' + not + ')');

    // set agent no save (this is the default).
    setNoSave();

    try {
      if (not instanceof ReceiveRequest)
        receiveRequest(from, (ReceiveRequest) not);
      else if (not instanceof BrowseRequest)
        browseRequest(from, (BrowseRequest) not);
      else if (not instanceof AcknowledgeRequest)
        acknowledgeRequest((AcknowledgeRequest) not);
      else if (not instanceof DenyRequest)
        denyRequest(from, (DenyRequest) not);
      else if (not instanceof AbortReceiveRequest)
        abortReceiveRequest(from, (AbortReceiveRequest) not);
      else if (not instanceof ExpiredNot)
        handleExpiredNot(from, (ExpiredNot) not);
      else if (not instanceof QueueDeliveryTimeNot)
        processDeliveryTime(from, (QueueDeliveryTimeNot) not);
      else
        super.react(from, not);
    } catch (MomException exc) {
      // MOM Exceptions are sent to the requester.
      if (logger.isLoggable(BasicLevel.WARN))
        logger.log(BasicLevel.WARN, "", exc);

      if (not instanceof AbstractRequestNot) {
        AbstractRequestNot req = (AbstractRequestNot) not;
        Channel.sendTo(from, new ExceptionReply(req, exc));
      }
    }
  }
  
  protected void agentSave() throws IOException {
    super.agentSave();
    arrivalState.save();
    deliveryTable.save();
  }

  /**
   * Removes all request that the expiration time is expired.
   * 
   * Be careful,this method is part of the Queue MBean interface, it can be called outside the
   * reactions of the engine thread. In order to avoid synchronization issues we should avoid
   * direct manipulation of requests list (see JORAM-372, JORAM-373).
   * @throws Exception 
   * @throws InterruptedException 
   */
  @Override
  public final void cleanWaitingRequest() throws Exception {
    AdminSyncNotification cmd = new AdminSyncNotification(AdminSyncNotification.CLEAN_WAITING_REQUEST, true);
    cmd.invoke(getId());
    return;
  }

  /**
   * Cleans the waiting request list.
   * Removes all request that the expiration time is less than the time
   * given in parameter.
   *
   * @param currentTime The current time.
   */
  protected void cleanWaitingRequest(long currentTime) {
    int index = 0;
    while (index < requests.size()) {
      if (! ((ReceiveRequest) requests.get(index)).isValid(currentTime)) {
        // Request expired or context closed: removing it
        requests.remove(index);
        // It's not really necessary to save its state, in case of failure
        // a similar work will be done at restart.
      } else {
        index++;
      }
    }
  }

  /**
   * Returns the number of waiting requests in the queue.
   *
   * @return The number of waiting requests.
   */
  @Override
  public final int getWaitingRequestCount() {
    if (requests != null) {
      // Be careful, do not call cleanWaitingRequest to avoid issue due to concurrent access to requests.
      // The number of pending requests may possibly be higher than reality because some may be outdated.
      // See #314361.
//      cleanWaitingRequest();
      return requests.size();
    }
    return 0;
  }

  /** <code>true</code> if the queue is currently handling a new received message. */
  protected transient boolean receiving = false;

  /** List holding the messages before delivery. */
  protected transient List<Message> messages;
  
  /** List of delayed messages */
  protected transient List<Message> delayed;
  
  /**
   * Returns the number of messages waiting for a delay.
   *
   * @return The number of messages waiting for a delay.
   */
  public final int getDelayedMessageCount() {
    return delayed.size();
  }

  public byte getType() {
    return DestinationConstants.QUEUE_TYPE;
  }

  /**
   * Removes all messages that the time-to-live is expired.
   * 
   * Be careful,this method is part of the Queue MBean interface, it can be called outside the
   * reactions of the engine thread. In order to avoid synchronization issues we should avoid
   * direct manipulation of messages list (see JORAM-372, JORAM-373).
   * @throws Exception 
   * @throws InterruptedException 
   */
  @Override
  public final void cleanPendingMessage() throws Exception {
    AdminSyncNotification cmd = new AdminSyncNotification(AdminSyncNotification.CLEAN_PENDING_MESSAGE, true);
    cmd.invoke(getId());
    return;
  }

  /**
   * Cleans the pending messages list. Removes all messages which expire before
   * the date given in parameter.
   * 
   * @param currentTime
   *          The current time.
   * @return A <code>DMQManager</code> which contains the expired messages.
   *         <code>null</code> if there wasn't any.
   */
  protected DMQManager cleanPendingMessage(long currentTime) {
    DMQManager dmqManager = null;
    
    if (nbExpirations == 0)
      return dmqManager;
    
    // Be careful, browsing the message list is expensive when there are many messages waiting.
    int index = 0;
    Message message = null;
    while (index < messages.size()) {
      message = (Message) messages.get(index);
      if (! message.isValid(currentTime)) {
        messages.remove(index);
        nbExpirations -= 1;

        if (dmqManager == null)
          dmqManager = new DMQManager(dmqId, getId());
        nbMsgsSentToDMQSinceCreation++;
        // JORAM-326: Saves the queue to avoid bad counter value at restart.
        if (strictCounters) setSave();
        message.delete();
        dmqManager.addDeadMessage(message.getFullMessage(), MessageErrorConstants.EXPIRED);

        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Removes expired message " + message.getId());
      } else {
        index++;
      }
    }
    return dmqManager;
  }

  /**
   * Returns the number of pending messages in the queue.
   *
   * @return The number of pending messages.
   */
  public final int getPendingMessageCount() {
    if (messages != null) {
      return messages.size();
    }
    return 0;
  }

//  /**
//   * This task allow to compute the average load of the queue.
//   */
//  transient QueueAverageLoadTask averageLoadTask = null;
//  
//  /**
//   * Returns the load averages for the last minute.
//   * @return the load averages for the last minute.
//   */
//  public float getAverageLoad1() {
//    return (averageLoadTask==null)?0:averageLoadTask.getAverageLoad1();
//  }
//
//  /**
//   * Returns the load averages for the past 5 minutes.
//   * @return the load averages for the past 5 minutes.
//   */
//  public float getAverageLoad5() {
//    return (averageLoadTask==null)?0:averageLoadTask.getAverageLoad5();
//  }
//  
//  /**
//   * Returns the load averages for the past 15 minutes.
//   * @return the load averages for the past 15 minutes.
//   */
//  public float getAverageLoad15() {
//    return (averageLoadTask==null)?0:averageLoadTask.getAverageLoad15();
//  }
//
//  class QueueAverageLoadTask extends AverageLoadTask {
//    Queue queue = null;
//    
//    public QueueAverageLoadTask(Timer timer, Queue queue) {
//      this.queue = queue;
//      start(timer);
//    }
//    
//    /**
//     * Returns the number of waiting messages in the engine.
//     * 
//     * @see fr.dyade.aaa.common.AverageLoadTask#countActiveTasks()
//     */
//    @Override
//    protected long countActiveTasks() {
//      return queue.getPendingMessageCount();
//    }
//  }

  /**
   * Returns the number of messages delivered and waiting for acknowledge.
   *
   * @return The number of messages delivered.
   */
  public final int getDeliveredMessageCount() {
    if (deliveryTable != null) {
      return deliveryTable.size();
    }
    return 0;
  }
  
  protected long nbMsgsDeniedSinceCreation = 0;
  
  /**
   * Returns the number of messages denied since creation time of this
   * destination.
   *
   * @return the number of messages delivered since creation time.
   */
  public final long getNbMsgsDeniedSinceCreation() {
    return nbMsgsDeniedSinceCreation;
  }

  public long getNbMsgsDeliverSinceCreation() {
    return nbMsgsDeliverSinceCreation - nbMsgsDeniedSinceCreation;
  }

  public long getNbMsgsReceiveSinceCreation() {
    return nbMsgsSentToDMQSinceCreation + nbMsgsDeliverSinceCreation + getPendingMessageCount() - nbMsgsDeniedSinceCreation;
  }

  /** nb Max of Message store in queue (-1 no limit). */
  protected int nbMaxMsg = -1;

  /**
   * Returns the maximum number of message for the destination.
   * If the limit is unset the method returns -1.
   *
   * @return the maximum number of message for subscription if set;
   *	     -1 otherwise.
   */
  public final int getNbMaxMsg() {
    return nbMaxMsg;
  }

  /**
   * Sets the maximum number of message for the destination.
   *
   * @param nbMaxMsg the maximum number of message (-1 set no limit).
   */
  public void setNbMaxMsg(int nbMaxMsg) {
    // state change, so save.
    setSave();
    this.nbMaxMsg = nbMaxMsg;
  }

  /**
   * Initializes the destination.
   * 
   * @param firstTime		true when first called by the factory
   */
  protected void initialize(boolean firstTime) throws Exception {
    cleanWaitingRequest(System.currentTimeMillis());

    receiving = false;

    //    averageLoadTask = new QueueAverageLoadTask(AgentServer.getTimer(), this);

    String arrivalStateTxName = ARRIVAL_STATE_PREFIX + getId().toString();
    String deliveryTableTxName = DELIVERY_TABLE_PREFIX + getId().toString();
    if (firstTime) {
      arrivalState = new QueueArrivalState(arrivalStateTxName);
      deliveryTable = new QueueDeliveryTable(deliveryTableTxName);
      messages = new Vector<Message>();
      delayed = new Vector<Message>();
      return;
    } else {
      arrivalState = QueueArrivalState.load(arrivalStateTxName);
      deliveryTable = QueueDeliveryTable.load(deliveryTableTxName);
      delayed = new Vector<Message>();
    }

    // Retrieving the persisted messages, if any.
    messages = Message.loadAll(getMsgTxPrefix().toString(), Integer.MAX_VALUE);
    long currentTime = System.currentTimeMillis();
    if (logmsg.isLoggable(BasicLevel.INFO))
      logmsg.log(BasicLevel.INFO, getName() + " [" + getAgentId() + "], start retrieves messages");

    for (int index=0; index < messages.size(); ) {
      Message persistedMsg = messages.get(index);
      if (logmsg.isLoggable(BasicLevel.INFO))
        logmsg.log(BasicLevel.INFO, getName() + ": retrieves message " + persistedMsg.getId() + " -> " + persistedMsg.getDeliveryTime());

      QueueDelivery queueDelivery = deliveryTable.get(persistedMsg.getId());
      if (queueDelivery == null) {
        if (persistedMsg.hasExpiration())
          nbExpirations += 1;

        if (persistedMsg.getDeliveryTime() > currentTime) {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG, getName() + ": schedule delayed message " + persistedMsg.getId());
          delayed.add(persistedMsg);
          // TODO (AF): Be careful, this way of handling timed messages is not scalable. We should maintain an
          // ordered list of timed messages, and set a timer for the first timeout (see Scheduler class).
          AgentServer.getTimer().schedule(new QueueDeliveryTimeTask(getId(), persistedMsg, false), new Date(persistedMsg.getDeliveryTime()));
          // Remove message from the list of messages to deliver.
          messages.remove(index);
          // Do not increment index.
          continue;
        } else {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG,
                       getName() + ": Adds message " + persistedMsg.getId() + " in the list of messages to deliver.");
        }
      } else {
        // The message has been delivered before stop.
        queueDelivery.setMessage(persistedMsg);
        if (isLocal(queueDelivery.getConsumerId())) {
          // The delivery is aborted.
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG, " -> deny " + persistedMsg.getId());
          deliveryTable.remove(persistedMsg.getId());
          if (persistedMsg.hasExpiration())
            nbExpirations += 1;
        } else {
          // The delivery is always active, remove message from the list of messages to deliver.
          messages.remove(index);
          // Do not increment index.
          continue;
        }
      }

      index += 1;
    }

    // Searches for bad deliveries (delivery null, or associated with a message not reloaded from persistence), then deletes them.
    // Note (AF): Should never happen!
    String[] msgids = deliveryTable.getMessageIds();
    for (String msgid : msgids) {
      QueueDelivery delivery = deliveryTable.get(msgid);
      if (delivery == null) {
        logger.log(BasicLevel.WARN, getName() + " [" + getAgentId() + "] removes null delivery for" + msgid + '.');
        deliveryTable.remove(msgid);
      } else if (delivery.getMessage() == null) {
        logger.log(BasicLevel.WARN, getName() + " [" + getAgentId() + "] removes delivery for" + msgid + ", message is null.");
        deliveryTable.remove(msgid);
      }
    }

    if (! messages.isEmpty()) {
      if (messages.get(0).getPriority() == messages.get(messages.size() -1).getPriority()) {
        samePriorities = true;
        priority = messages.get(0).getPriority();
      }
    }
    
    if (logmsg.isLoggable(BasicLevel.INFO))
      logmsg.log(BasicLevel.INFO, getName() + ", end retrieves messages -> " + (System.currentTimeMillis() - currentTime));
  }
  
  /**
   * Finalizes the destination before it is garbaged.
   * 
   * @param last true if the destination is deleted
   */
  protected void finalize(boolean last) {
    setSave();
//    averageLoadTask.cancel();
//    averageLoadTask = null;
  }

  /**
   * Returns a string representation of this destination.
   * @return a string representation of this destination.
   */
  public String toString() {
    return "Queue:" + getId().toString();
  }

  long hprod, hcons;
  int pload = -1;
  int cload = -1;

  /**
   * wake up, and cleans the queue.
   */
  public void wakeUpNot(WakeUpNot not) {
    long current = System.currentTimeMillis();
    cleanWaitingRequest(current);
    // Cleaning the possibly expired messages.
    DMQManager dmqManager = cleanPendingMessage(current);
    // If needed, sending the dead messages to the DMQ:
    if (dmqManager != null) {
      setSave();
      dmqManager.sendToDMQ();
    }

    long prod = getNbMsgsReceiveSinceCreation();
    long cons = getNbMsgsDeliverSinceCreation();
//    if ((hprod == 0) && (hcons == 0)) {
//      hprod = prod;
//      hcons = cons;
//    }
    pload = (pload + 2*(int)((1000L*(prod-hprod))/getPeriod()))/3;
    cload = (cload + 2*(int)((1000L*(cons-hcons))/getPeriod()))/3;
    hprod = prod;
    hcons = cons;
  }
  
  /**
   * Return the average producer's load during last moments.
   */
  public long getProducerLoad() {
    return pload;
  }

  /**
   * Return the average consumer's load during last moments.
   */
  public long getConsumerLoad() {
    return cload;
  }

  /**
   * This method allows to exclude some JMX attribute of getJMXStatistics method.
   * It excludes.
   * 
   * @param attrName name of attribute to test.
   * @return true if the attribute is a valid one.
   */
  protected boolean isValidJMXAttribute(String attrName) {
    if ("Messages".equals(attrName))
      return false;
    return super.isValidJMXAttribute(attrName);
  }

  /**
   * Method implementing the reaction to a <code>ReceiveRequest</code>
   * instance, requesting a message.
   * <p>
   * This method stores the request and launches a delivery sequence.
   *
   * @exception AccessException  If the sender is not a reader.
   */
  protected void receiveRequest(AgentId from, ReceiveRequest not) throws AccessException {
    // If client is not a reader, sending an exception.
    if (! isReader(from))
      throw new AccessException("READ right not granted");

    String[] toAck = not.getMessageIds();
    if (toAck != null) {
      for (int i = 0; i < toAck.length; i++) {
        acknowledge(toAck[i]);
      }
    }

    long current = System.currentTimeMillis();
    cleanWaitingRequest(current);
    // Storing the request:
    not.requester = from;
    not.setExpiration(current);
    if (not.isPersistent()) {
      // TODO: the requests could be saved externally.
      // state change, so save.
      setSave();
    }
    requests.add(not);

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, " -> requests count = " + requests.size());

    // Launching a delivery sequence for this request:
    int reqIndex = requests.size() - 1;
    deliverMessages(reqIndex);

    // If the request has not been answered and if it is an immediate
    // delivery request, sending a null:
    if ((requests.size() - 1) == reqIndex && not.getTimeOut() == -1) {
      requests.remove(reqIndex);
      QueueMsgReply reply = new QueueMsgReply(not);
      if (isLocal(from)) {
        reply.setPersistent(false);
      }
      forward(from, reply);

      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Receive answered by a null.");
    }
  }

  /**
   * Method implementing the queue reaction to a <code>BrowseRequest</code>
   * instance, requesting an enumeration of the messages on the queue.
   * <p>
   * The method sends a <code>BrowseReply</code> back to the client. Expired
   * messages are sent to the DMQ.
   *
   * @exception AccessException  If the requester is not a reader.
   */
  protected void browseRequest(AgentId from, BrowseRequest not) throws AccessException {
    // If client is not a reader, sending an exception.
    if (! isReader(from))
      throw new AccessException("READ right not granted");

    // Building the reply:
    BrowseReply rep = new BrowseReply(not);

    // Cleaning the possibly expired messages.
    DMQManager dmqManager = cleanPendingMessage(System.currentTimeMillis());
    // Adding the deliverable messages to it:
    Message message;
    for (int i = 0; i < messages.size(); i++) {
      message = (Message) messages.get(i);
      if (Selector.matches(message.getHeaderMessage(), not.getSelector())) {
        // Matching selector: adding the message:
        rep.addMessage(message.getFullMessage());
      }
    }
    // TODO (AF): May be we have to handle the delayed messages?
    
    // Sending the dead messages to the DMQ, if needed:
    if (dmqManager != null)
      dmqManager.sendToDMQ();

    // Delivering the reply:
    forward(from, rep);

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Request answered.");
  }

  /**
   * Method implementing the reaction to an <code>AcknowledgeRequest</code>
   * instance, requesting messages to be acknowledged.
   */
  protected void acknowledgeRequest(AcknowledgeRequest not) {
    for (Enumeration<String> ids = not.getIds(); ids.hasMoreElements();) {
      String msgId = ids.nextElement();
      acknowledge(msgId);
    }
  }

  private void acknowledge(String msgId) {
    if (logmsg.isLoggable(BasicLevel.INFO))
      logmsg.log(BasicLevel.INFO, getName() + ": acknowledges message " + msgId);

    QueueDelivery queueDelivery = deliveryTable.remove(msgId);
    if ((queueDelivery != null) && (queueDelivery.getMessage() != null)) {
      // The DeliveryTable is saved outside the Queue agent
      /*
      if (queueDelivery.getMessage().isPersistent()) {
        // state change, so save.
        setSave();
      }*/

      queueDelivery.getMessage().delete();

      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Message " + msgId + " acknowledged.");
    } else {
      if ((logger.isLoggable(BasicLevel.WARN) || logmsg.isLoggable(BasicLevel.WARN))) {
        String str = getName() + ": message " + msgId + " not found for acknowledgement.";
        logger.log(BasicLevel.WARN, str);
        logmsg.log(BasicLevel.WARN, str);
      }
    }
  }

  /**
   * Method implementing the reaction to a <code>DenyRequest</code>
   * instance, requesting messages to be denied.
   * <p>
   * This method denies the messages and launches a delivery sequence.
   * Messages considered as undeliverable are sent to the DMQ.
   */
  protected void denyRequest(AgentId from, DenyRequest not) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.DenyRequest(" + from + ',' + not + ')');
    
    DMQManager dmqManager = null;

    Enumeration<String> ids = not.getIds();
    if (! ids.hasMoreElements()) {
      // If the deny request is empty, the denying is a contextual one: it requests the denying of all the messages
      // consumed by the denier in the denying context:
      String[] msgids = deliveryTable.getMessageIds();
      for (String msgid : msgids) {
        // Delivery should never be null
        QueueDelivery delivery = deliveryTable.get(msgid);

        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, " -> deny msg " + msgid + "(consId = " + delivery.getConsumerId() + ')');

        // If the current message has been consumed by the denier in the same context: denying it.
        if (delivery.getConsumerId().equals(from) && delivery.getContextId() == not.getClientContext()) {
          dmqManager = denyMessage(msgid, not.isRedelivered(), dmqManager);
        }
      }
    } else {
      // For a non empty request, browsing the denied messages:
      while (ids.hasMoreElements()) {
        String msgid = ids.nextElement();
        dmqManager = denyMessage(msgid, not.isRedelivered(), dmqManager);
      }
    }
    // Sending the dead messages to the DMQ, if needed:
    if (dmqManager != null)
      dmqManager.sendToDMQ();

    // Launching a delivery sequence:
    deliverMessages(0);
  }

  /**
   * Denies the specified message.
   * 
   * @param msgid The identifier of message to deny.
   * @param redelivered True if the message has been denied from user.
   * @param dmqManager The DMQManager if it exists, null otherwise.
   * @return The DMQManager if it exists, null otherwise.
   */
  private DMQManager denyMessage(String msgid, boolean redelivered, DMQManager dmqManager) {
    QueueDelivery delivery = deliveryTable.remove(msgid);

    // Message may have already been denied. For example, a proxy may deny a message twice, first when
    // detecting a connection failure (and in that case it sends a contextual denying), then when receiving
    // the message from the queue - and in that case it also sends an individual denying.
    if (delivery == null) {
      if ((logger.isLoggable(BasicLevel.WARN) || logmsg.isLoggable(BasicLevel.WARN)))
        logmsg(BasicLevel.WARN, BasicLevel.WARN, getName() + ": already denied message " + msgid);
      return dmqManager;
    }

    nbMsgsDeniedSinceCreation += 1;
    // JORAM-326: Saves the queue to avoid bad counter value at restart.
    if (strictCounters) setSave();

    Message message = delivery.getMessage();
    if (message == null) {
      if ((logger.isLoggable(BasicLevel.WARN) || logmsg.isLoggable(BasicLevel.WARN)))
        logmsg(BasicLevel.WARN, BasicLevel.WARN, getName() + ": message " + msgid + " not found for deny.");
      return dmqManager;
    }

    // Note (AF):This code complies with the JMS specification: a message should only be marked redelivered if it
    //has actually been delivered to the JMS client and its rejection is due to the processing by the client.
    // However, in the case where a message is repeatedly rejected for an error independent of the client, this prevents
    // the activation of prevention mechanisms for delivery loop (redelivery delay, threshold delivery).
    if (redelivered)
      message.setRedelivered();
    else if (!lenientUndeliverable)
      message.decDeliveryCount();
    
    // TODO (AF): Enable delayed/threshold mechanisms for messages not marked redelivered.
    // /!\ Be careful, this message is maybe undeliverable!!
    // isRedelivered => DeliveryCount() > 0
    if (getRedeliveryDelay() > 0 && (message.isRedelivered() || lenientRedeliveryDelay)) {
      message.setDeliveryTime(System.currentTimeMillis() + (getRedeliveryDelay() *1000L));
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Queue.DenyRequest: setDeliveryTime " + message.getDeliveryTime());
    }

    if ((logger.isLoggable(BasicLevel.DEBUG) || logmsg.isLoggable(BasicLevel.INFO)))
      logmsg(BasicLevel.DEBUG, BasicLevel.INFO, getName() + ": denies message " + msgid);

    // If message considered as undeliverable, adding it to the list of dead messages:
    if (isUndeliverable(message)) {
      message.delete();
      if (dmqManager == null)
        dmqManager = new DMQManager(dmqId, getId());
      nbMsgsSentToDMQSinceCreation++;
      // JORAM-326: If needed setSave() is already done above.
      dmqManager.addDeadMessage(message.getFullMessage(), MessageErrorConstants.UNDELIVERABLE);
    } else {
      //Reinserts the message, either in the list of messages to be delivered, or in the list of messages to be delivered after a delay.
      try {
        if (message.getDeliveryTime() > 0) {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG, "Queue.DenyRequest: scheduleDeliveryTimeMessage " + message.getId() + ", reDeliveryDelay = " + getRedeliveryDelay());
          // TODO (AF): Should test for queue overflow.
          addDeliveryTimeMessage(message, false, true);
        } else {
          // Else, putting the message back into the deliverables list:
          if (addMessage(message, false)) {
            // Persisting the message.
            message.saveHeader();
            message.releaseFullMessage();
            if (logger.isLoggable(BasicLevel.DEBUG))
              logger.log(BasicLevel.DEBUG, "Message " + message.getId() + " stored.");
          }
        }
      } catch (AccessException e) { /* never happens */
        // TODO (AF): Deletes the message and logs a message.
      }
    }       
    if ((logger.isLoggable(BasicLevel.DEBUG) || logmsg.isLoggable(BasicLevel.INFO)))
      logmsg(BasicLevel.DEBUG, BasicLevel.INFO, getName() + ": denies message " + msgid);
    
    return dmqManager;
  }

  /**
   * Logs message either in logmsg if configured, or in logger.
   * 
   * @param loggerLevel
   * @param logmsgLevel
   * @param logMessage
   */
  private void logmsg(Level loggerLevel, Level logmsgLevel, String logMessage) {
    if (logmsg.isLoggable(logmsgLevel))
      logmsg.log(logmsgLevel, logMessage);
    else
      logger.log(loggerLevel, logMessage);
  }
  
  protected void abortReceiveRequest(AgentId from, AbortReceiveRequest not) {
    for (int i = 0; i < requests.size(); i++) {
      ReceiveRequest request = (ReceiveRequest) requests.get(i);
      if (request.requester.equals(from) &&
          request.getClientContext() == not.getClientContext() &&
          request.getRequestId() == not.getAbortedRequestId()) {
        if (not.isPersistent()) {
          // state change, so save.
          setSave();
        }
        requests.remove(i);
        break;
      }
    }
  }

  /**
   * @see org.objectweb.joram.mom.dest.Destination#handleAdminRequestNot(fr.dyade.aaa.agent.AgentId, org.objectweb.joram.mom.notifications.FwdAdminRequestNot)
   */
  public void handleAdminRequestNot(AgentId from, FwdAdminRequestNot not) {
    AdminRequest adminRequest = not.getRequest();
    
    if (adminRequest instanceof GetQueueMessageIds) {
      getQueueMessageIds(not.getReplyTo(),
                         not.getRequestMsgId(),
                         not.getReplyMsgId());
    } else if (adminRequest instanceof GetQueueMessage) {
      getQueueMessage((GetQueueMessage)adminRequest,
                      not.getReplyTo(),
                      not.getRequestMsgId(),
                      not.getReplyMsgId());
    } else if (adminRequest instanceof DeleteQueueMessage) {
      deleteQueueMessage((DeleteQueueMessage)adminRequest,
                         not.getReplyTo(),
                         not.getRequestMsgId(),
                         not.getReplyMsgId());
    } else if (adminRequest instanceof ClearQueue) {
      clearQueue(not.getReplyTo(),
                 not.getRequestMsgId(),
                 not.getReplyMsgId());
    } else if (adminRequest instanceof GetNbMaxMsgRequest) {
      replyToTopic(new GetNumberReply(getNbMaxMsg()),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof GetPendingMessages) {
      // Cleaning of the possibly expired messages.
      DMQManager dmqManager = cleanPendingMessage(System.currentTimeMillis());
      // Sending the dead messages to the DMQ, if needed:
      if (dmqManager != null) dmqManager.sendToDMQ();
      
      replyToTopic(new GetNumberReply(getPendingMessageCount()),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof GetPendingRequests) {
      // Cleaning of the possibly expired requests.
      cleanWaitingRequest(System.currentTimeMillis());
      replyToTopic(new GetNumberReply(getWaitingRequestCount()),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof GetDMQSettingsRequest) {
      replyToTopic(new GetDMQSettingsReply((dmqId != null)?dmqId.toString():null, threshold),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof SetThresholdRequest) {
      setSave(); // state change, so save.
      threshold = ((SetThresholdRequest) adminRequest).getThreshold();
      
      replyToTopic(new AdminReply(true, null),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof SetNbMaxMsgRequest) {
      setSave(); // state change, so save.
      nbMaxMsg = ((SetNbMaxMsgRequest) adminRequest).getNbMaxMsg();

      replyToTopic(new AdminReply(true, null),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof SetSyncExceptionOnFullDestRequest) {
      setSave(); // state change, so save.
      syncExceptionOnFullDest = ((SetSyncExceptionOnFullDestRequest) adminRequest).isSyncExceptionOnFullDest();

      replyToTopic(new AdminReply(true, null),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else if (adminRequest instanceof GetDeliveredMessages) {
      // TODO (AF): Should be getNbMsgsDeliverSinceCreation?
      replyToTopic(new GetNumberReply((int)nbMsgsDeliverSinceCreation),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
    } else {
    	super.handleAdminRequestNot(from, not);
    }
  }

  private void getQueueMessageIds(AgentId replyTo,
                                  String requestMsgId,
                                  String replyMsgId) {
    String[] res = new String[messages.size()];
    for (int i = 0; i < messages.size(); i++) {
      Message msg = (Message) messages.get(i);
      res[i] = msg.getId();
    }
    replyToTopic(new GetQueueMessageIdsRep(res), replyTo, requestMsgId, replyMsgId);
  }

  private void getQueueMessage(GetQueueMessage request,
                               AgentId replyTo,
                               String requestMsgId,
                               String replyMsgId) {
    Message message = null;

    for (int i = 0; i < messages.size(); i++) {
      message = (Message) messages.get(i);
      if (message.getId().equals(request.getMessageId())) break;
      message = null;
    }

    if (message != null) {
      GetQueueMessageRep reply = null;
      if (request.getFullMessage()) {
        reply = new GetQueueMessageRep(message.getFullMessage());
      } else {
        reply = new GetQueueMessageRep(message.getHeaderMessage());
      }
      replyToTopic(reply, replyTo, requestMsgId, replyMsgId);
    } else {
      replyToTopic(new AdminReply(false, "Unknown message " + request.getMessageId()), replyTo, requestMsgId,
          replyMsgId);
    }
  }

  private void deleteQueueMessage(DeleteQueueMessage request,
                                  AgentId replyTo,
                                  String requestMsgId,
                                  String replyMsgId) {
    // Sends remaining messages to DMQ
    for (int i = 0; i < messages.size(); i++) {
      Message message = (Message) messages.get(i);
      if (message.getId().equals(request.getMessageId())) {
        messages.remove(i);
        if ((nbExpirations > 0 ) && message.hasExpiration())
          nbExpirations -= 1;
        message.delete();
        DMQManager dmqManager = new DMQManager(dmqId, getId());
        nbMsgsSentToDMQSinceCreation++;
        // JORAM-326: Saves the queue to avoid bad counter value at restart.
        if (strictCounters) setSave();
        dmqManager.addDeadMessage(message.getFullMessage(), MessageErrorConstants.ADMIN_DELETED);
        dmqManager.sendToDMQ();
        break;
      }
    }
    replyToTopic(new AdminReply(true, null), replyTo, requestMsgId, replyMsgId);
  }

  /**
   * Removes all pending messages.
   */
  @Override
  public void clear() {
    sendTo(getId(), new FwdAdminRequestNot(new ClearQueue(), null, null, null));
  }
  
  private void clearQueue(AgentId replyTo,
                          String requestMsgId,
                          String replyMsgId) {
    if (messages.size() > 0) {
      DMQManager dmqManager = new DMQManager(dmqId, getId());
      for (int i = 0; i < messages.size(); i++) {
        Message message = (Message) messages.get(i);
        message.delete();
        nbMsgsSentToDMQSinceCreation++;
        // JORAM-326: Saves the queue to avoid bad counter value at restart.
        if (strictCounters) setSave();
        dmqManager.addDeadMessage(message.getFullMessage(), MessageErrorConstants.ADMIN_DELETED);
      }
      dmqManager.sendToDMQ();
      messages.clear();
      nbExpirations = 0;
    }
    if (replyTo != null)
      replyToTopic(new AdminReply(true, null), replyTo, requestMsgId, replyMsgId);
  }

  /**
   * Method specifically processing a <code>SetRightRequest</code> instance.
   * <p>
   * When a reader is removed, and receive requests of this reader are still
   * on the queue, they are replied to by an <code>ExceptionReply</code>.
   */
  protected void doRightRequest(AgentId user, int right) {
    // If the request does not unset a reader, doing nothing.
    if (right != -READ) return;

    ReceiveRequest request;

    if (user == null) {
      // Free reading right has been removed, reject the non readers requests.
      for (int i = 0; i < requests.size(); i++) {
        request = (ReceiveRequest) requests.get(i);
        if (! isReader(request.requester)) {
          forward(request.requester,
                  new ExceptionReply(request, new AccessException("Free READ access removed")));
          setSave(); // state change, so save.
          requests.remove(i);
          i--;
        }
      }
    } else {
      // Reading right of a given user has been removed; replying to its requests.
      for (int i = 0; i < requests.size(); i++) {
        request = (ReceiveRequest) requests.get(i);
        if (user.equals(request.requester)) {
          forward(request.requester,
                  new ExceptionReply(request, new AccessException("READ right removed")));
          setSave(); // state change, so save.
          requests.remove(i);
          i--;
        }
      }
    }
  }

  /**
   * Method specifically processing a <code>ClientMessages</code> instance.
   * <p>
   * This method stores the messages and launches a delivery sequence.
   * 
   * This method is used when ClientMessages comes from a JMS client and/or LB notification in
   * ClusterQueue (May be we should use addClienMessages). It is also used in old deprecated
   * JMS bridge.
   */
  protected void doClientMessages(AgentId from, ClientMessages not, boolean throwsExceptionOnFullDest) throws AccessException {
    receiving = true;
    ClientMessages cm = null;
    
    // interceptors
    if (interceptorsAvailable()) {
    	// new client message
    	cm = new ClientMessages(not.getClientContext(), not.getRequestId());
    	cm.setAsyncSend(not.getAsyncSend());
    	cm.setDMQId(not.getDMQId());
    	
    	for (Iterator<org.objectweb.joram.shared.messages.Message> msgs = not.getMessages().iterator(); msgs.hasNext();) {
    		org.objectweb.joram.shared.messages.Message message = msgs.next();
    		// set the destination name
    		message.setProperty("JoramDestinationName", getName());
        
        // JORAM-323: If it is a temporary destination there is no need to persistent messages.
        if (temporary) {
          message.persistent = false;
        }

    		// interceptors process
    		org.objectweb.joram.shared.messages.Message m = processInterceptors(message);
    		if (m == null) {
    			// send message to the DMQ
    			DMQManager dmqManager = new DMQManager(dmqId, getId());
    			nbMsgsSentToDMQSinceCreation++;
          // JORAM-326: Saves the queue to avoid bad counter value at restart.
          if (strictCounters) setSave();
    			dmqManager.addDeadMessage(message, MessageErrorConstants.INTERCEPTORS);
    			dmqManager.sendToDMQ();
    		} else {
    			// add message to the client message 
    			cm.addMessage(m);
    		}
    	}
    	// test client message size.
    	if (cm.getMessageCount() == 0) {
    		receiving = false;
    		return;
    	}
    } else {
    	cm = not;
    }
    
    // pre process the client message
    ClientMessages clientMsgs = preProcess(from, cm);
   
    long currentTime = System.currentTimeMillis();
    long deliveryTime = currentTime;
    if (deliveryDelay > 0)
      deliveryTime += deliveryDelay;
    
    if (clientMsgs != null) {
      Message msg;
      // Storing each received message:
      for (Iterator<org.objectweb.joram.shared.messages.Message> msgs = clientMsgs.getMessages().iterator(); msgs.hasNext();) {
        org.objectweb.joram.shared.messages.Message sharedMsg = msgs.next();
        msg = new Message(sharedMsg);
        msg.order = arrivalState.getAndIncrementArrivalCount(msg.isPersistent());

        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Queue.doClientMessages() -> " + msg.getId() + ',' + msg.order);

        if ((deliveryTime > currentTime) && (deliveryTime > sharedMsg.deliveryTime))
          sharedMsg.deliveryTime = deliveryTime;
        
        if (sharedMsg.deliveryTime > currentTime) {
          addDeliveryTimeMessage(msg, throwsExceptionOnFullDest, false);
        } else {
          storeMessage(msg, throwsExceptionOnFullDest);
          
          // ArrivalState is saved outside the Queue agent
          // if (msg.isPersistent()) setSave();
        }
      }
    }

    // Launching a delivery sequence:
    deliverMessages(0);

    if (clientMsgs != null)
      postProcess(clientMsgs);

    receiving = false;
  }

  private static final String getIdString(Message msg) {
    String jmsCorrelationId = msg.getCorrelationId();
    Object joramCorrelationId = msg.getSharedMsg().getProperty(msg.getSharedMsg().CORRELATION_ID);
    
    if ((jmsCorrelationId == null) && (joramCorrelationId == null))
      return msg.getId();
    
    if (jmsCorrelationId == null) jmsCorrelationId = "";
    if (joramCorrelationId == null) joramCorrelationId = "";
    
    return msg.getId() + " [" + jmsCorrelationId + '/' + joramCorrelationId + "]";
  }
  
  /**
   * Adds the specified message to the list of delayed message.
   * @param msg                       The message to add.
   * @param throwsExceptionOnFullDest if true, can throws an exception if destination is full.
   * @param isHeader                  If true, only save the message header (the message has already been saved).
   * @throws AccessException
   */
  void addDeliveryTimeMessage(Message msg, boolean throwsExceptionOnFullDest, boolean isHeader) throws AccessException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.addDeliveryTimeMessage(" + msg + ')');
    
    if (logmsg.isLoggable(BasicLevel.INFO))
      logmsg.log(BasicLevel.INFO, getName() + ": gets new delayed message " + getIdString(msg) + ", " + msg.order);

    // queue is full
    if (nbMaxMsg > -1 && nbMaxMsg <= (messages.size() + deliveryTable.size() + getDelayedMessageCount())) {
      if (throwsExceptionOnFullDest && isSyncExceptionOnFullDest()) {
        if (logger.isLoggable(BasicLevel.INFO))
          logger.log(BasicLevel.INFO, "addDeliveryTimeMessage " + msg.getId() + " throws Exception: The queue \"" + getName() + "\" is full (syncExceptionOnFullDest).");
        throw new AccessException("The queue \"" + getName() + "\" is full.");
      }
      DMQManager dmqManager = new DMQManager(dmqId, getId());
      nbMsgsSentToDMQSinceCreation++;
      // JORAM-326: Saves the queue to avoid bad counter value at restart.
      if (strictCounters) setSave();
      dmqManager.addDeadMessage(msg.getFullMessage(), MessageErrorConstants.QUEUE_FULL);
      dmqManager.sendToDMQ();
      return;
    }
    
    if (isHeader) {
      msg.saveHeader();
      msg.releaseFullMessage();
    } else {
      if (msg.isPersistent()) {
        // Persisting the message.
        setMsgTxName(msg);
        msg.save();
        msg.releaseFullMessage();
      }
    }
    
    delayed.add(msg);
    
    //schedule
    AgentServer.getTimer().schedule(new QueueDeliveryTimeTask(getId(), msg, throwsExceptionOnFullDest), new Date(msg.getDeliveryTime()));
  }
  
  void processDeliveryTime(AgentId from, QueueDeliveryTimeNot not) throws AccessException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.processDeliveryTime(" + from + ", " + not + ')');


    if (not.msg != null) {
      if (logmsg.isLoggable(BasicLevel.INFO))
        logmsg.log(BasicLevel.INFO, getName() + ": adds new delayed message " + not.msg.getId() + ", " + not.msg.order);
      
      // Adds a message in the list of messages to deliver.
      addMessage(not.msg, not.throwsExceptionOnFullDest);
      delayed.remove(not.msg);
    }
    
    // Launching a delivery sequence:
    deliverMessages(0);

    if (not.msg != null) {
      ClientMessages clientMsgs = new ClientMessages();
      clientMsgs.addMessage(not.msg.getSharedMsg());
      postProcess(clientMsgs);
    }
  }

  /**
   * Method specifically processing an <code>UnknownAgent</code> instance.
   * <p>
   * The specific processing is done when a <code>QueueMsgReply</code> was 
   * sent to a requester which does not exist anymore. In that case, the
   * messages sent to this requester and not yet acknowledged are marked as
   * "denied" for delivery to an other requester, and a new delivery sequence
   * is launched. Messages considered as undeliverable are removed and sent to
   * the DMQ.
   */ 
  protected void doUnknownAgent(UnknownAgent uA) {
    // If the notification is not a delivery, doing nothing. 
    if (! (uA.not instanceof QueueMsgReply))
      return;

    DMQManager dmqManager = null;
    // Note (AF): We could also use the list of messages from QueueMsgReply
    AgentId consumerId = uA.agent;
    int contextId = ((QueueMsgReply) uA.not).getClientContext();
    for (org.objectweb.joram.shared.messages.Message msg : ((QueueMsgReply) uA.not).getMessages()) {
      QueueDelivery delivery = deliveryTable.get(msg.id);
      // Delivered message has been delivered to the unknown client denying it.
      if ((delivery.getConsumerId().equals(consumerId)) && (delivery.getContextId() == contextId )) {
        dmqManager = denyMessage(msg.id, false, dmqManager);
      } else {
        if ((logger.isLoggable(BasicLevel.WARN) || logmsg.isLoggable(BasicLevel.WARN)))
          logmsg(BasicLevel.WARN, BasicLevel.WARN,
                 getName() + ": message " + msg.id + " not adequate for deny -> "+ consumerId + ", " + contextId);
      }
    }
    // Sending dead messages to the DMQ, if needed:
    if (dmqManager != null)
      dmqManager.sendToDMQ();

    // Launching a delivery sequence:
    deliverMessages(0);
  }

  /**
   * Method specifically processing a
   * <code>fr.dyade.aaa.agent.DeleteNot</code> instance.
   * <p>
   * <code>ExceptionReply</code> replies are sent to the pending receivers,
   * and the remaining messages are sent to the DMQ and deleted.
   */
  protected void doDeleteNot(DeleteNot not) {
    // Building the exception to send to the pending receivers:
    DestinationException exc = new DestinationException("Queue " + getId() + " is deleted.");
    ReceiveRequest rec;
    ExceptionReply excRep;
    // Sending it to the pending receivers:
    cleanWaitingRequest(System.currentTimeMillis());
    for (int i = 0; i < requests.size(); i++) {
      rec = (ReceiveRequest) requests.get(i);

      excRep = new ExceptionReply(rec, exc);
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG,
                   "Requester " + rec.requester + " notified of the queue deletion.");
      forward(rec.requester, excRep);
    }

    // JORAM-323: If it is a temporary destination do not send message to DMQ.
    if (temporary && (! AgentServer.getBoolean(DestinationConstants.TMPQUEUE_DMQ_ON_DELETE))) {
      logger.log(BasicLevel.INFO, 
                 "Removes temporary queue (" + getName() + ") and deletes " + messages.size() + " remaining messages.");
    } else {
      // Sending the remaining messages to the DMQ, if needed:
      if (! messages.isEmpty()) {
        Message message;
        DMQManager dmqManager = new DMQManager(dmqId, getId());
        while (! messages.isEmpty()) {
          message = (Message) messages.remove(0);
          message.delete();
          nbMsgsSentToDMQSinceCreation++;
          dmqManager.addDeadMessage(message.getFullMessage(), MessageErrorConstants.DELETED_DEST);
        }
        dmqManager.sendToDMQ();
      }
    }

    // Deleting the messages:
    Message.deleteAll(getMsgTxPrefix().toString());
  }

  transient StringBuffer msgTxPrefix = null;
  transient int msgTxPrefixLength = 0;

  protected final StringBuffer getMsgTxPrefix() {
    if (msgTxPrefix == null) {
      msgTxPrefix = new StringBuffer(18).append('M').append(getId().toString()).append('_');
      msgTxPrefixLength = msgTxPrefix.length();
    }
    return msgTxPrefix;
  }

  protected final void setMsgTxName(Message msg) {
    if (msg.getTxName() == null) {
      msg.setTxName(getMsgTxPrefix().append(msg.order).toString());
      msgTxPrefix.setLength(msgTxPrefixLength);
    }
  }

  /**
   * Actually stores a message in the deliverables list.
   * 
   * @param msg The message to store.
   * @param throwsExceptionOnFullDest true, can throws an exception on sending message on full destination
   * @throws AccessException
   */
  protected final void storeMessage(Message msg, boolean throwsExceptionOnFullDest) throws AccessException {
    if (addMessage(msg, throwsExceptionOnFullDest)) {
      if (logmsg.isLoggable(BasicLevel.INFO))
        logmsg.log(BasicLevel.INFO, getName() + ": adds new message " + getIdString(msg) + ", " + msg.order);

      if (msg.isPersistent()) {
        // Persisting the message.
        setMsgTxName(msg);
        msg.save();
        msg.releaseFullMessage();
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Message " + msg.getId() + " stored.");
    }
  }

  /** if true, throws an exception on sending message on full destination. */
  private boolean syncExceptionOnFullDest = false;
  
  /**
   * @return the syncExceptionOnFullDest
   */
  public boolean isSyncExceptionOnFullDest() {
    return syncExceptionOnFullDest;
  }

  /**
   * @param syncExceptionOnFullDest the syncExceptionOnFullDest to set
   */
  public void setSyncExceptionOnFullDest(boolean syncExceptionOnFullDest) {
    this.syncExceptionOnFullDest = syncExceptionOnFullDest;
  }

  /**
   * Adds a message in the list of messages to deliver.
   * This method take care of the message priority if needed.
   * 
   * @param message the message to add.
   * @param throwsExceptionOnFullDest true, can throws an exception on sending message on full destination
   * @return true if the message has been added. false if the queue is full.
   * @throws AccessException If syncExceptionOnFullDest and the queue isFull
   */
  protected final boolean addMessage(Message message, boolean throwsExceptionOnFullDest) throws AccessException {
    if (nbMaxMsg > -1 && nbMaxMsg <= (messages.size() + deliveryTable.size() + getDelayedMessageCount())) {
      if (throwsExceptionOnFullDest && isSyncExceptionOnFullDest()) {
        if (logger.isLoggable(BasicLevel.INFO))
          logger.log(BasicLevel.INFO, "addMessage " + message.getId() + " throws Exception: The queue \"" + getName() + "\" is full (syncExceptionOnFullDest).");
        throw new AccessException("The queue \"" + getName() + "\" is full.");
      }
      
      DMQManager dmqManager = new DMQManager(dmqId, getId());
      nbMsgsSentToDMQSinceCreation++;
      // JORAM-326: Saves the queue to avoid bad counter value at restart.
      if (strictCounters) setSave();
      dmqManager.addDeadMessage(message.getFullMessage(), MessageErrorConstants.QUEUE_FULL);
      dmqManager.sendToDMQ();
      return false;
    }

    if (messages.isEmpty()) {
      samePriorities = true;
      priority = message.getPriority();
      if (message.hasExpiration())
        nbExpirations = 1;
      else
        nbExpirations = 0;
    } else {
      if (samePriorities && priority != message.getPriority())
        samePriorities = false;
      if (message.hasExpiration())
        nbExpirations += 1;
    }

    if (samePriorities) {
      // Constant priorities: no need to insert the message according to its priority.
      if (receiving) {
        // Message being received: adding it at the end of the queue.
        messages.add(message);
      } else {
        // Denying or recovery: adding the message according to its original arrival order.
        int i = 0;
        // TODO (AF): May be it is not the best way to browse the message list.
        for (Iterator<Message> ite = messages.iterator(); ite.hasNext();) {
          if ((ite.next()).order > message.order) break;
          i++;
        }
        messages.add(i, message);
      }
    } else {
      // Non constant priorities: inserting the message according to its priority.
      Message currentMsg;
      int currentP;
      long currentO;
      int i = 0;
      for (Iterator<Message> ite = messages.iterator(); ite.hasNext();) {
        currentMsg = ite.next();
        currentP = currentMsg.getPriority();
        currentO = currentMsg.order;

        if (! receiving && currentP == message.getPriority()) {
          // Message denied or recovered, priorities are equal: inserting the
          // message according to its original arrival order.
          if (currentO > message.order) break;
        } else if (currentP < message.getPriority()) {
          // Current priority lower than the message to store: inserting it.
          break;
        }
        i++;
      }
      messages.add(i, message);
    }
    return true;
  }

  /**
   * Get a client message contain <code>nb</code> messages.
   * Only used in ClusterQueue.
   *  
   * @param nb        number of messages returned in ClientMessage.
   * @param selector  jms selector
   * @param remove    delete all messages returned if true
   * @return ClientMessages (contains nb Messages)
   */
  protected ClientMessages getClientMessages(int nb, String selector, boolean remove) {   
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.getClientMessages(" + nb + ',' + selector + ',' + remove + ')');

    ClientMessages cm = null ;
    List<Message> lsMessages = getMessages(nb, selector, remove);
    if (lsMessages.size() > 0) {
      cm = new ClientMessages();
      Iterator<Message> itMessages = lsMessages.iterator();
      while (itMessages.hasNext()) {
        cm.addMessage(itMessages.next().getFullMessage());
      }
    }
    return cm;
  }

  /**
   * List of message to be removed from messages vector.
   * No message.delete() call.
   * 
   * @param msgIds  List of message id.
   */
//  protected void removeMessages(List msgIds) {
//  	if (logger.isLoggable(BasicLevel.DEBUG))
//      logger.log(BasicLevel.DEBUG, "Queue.removeMessages(" + msgIds + ')');
//    String id = null;
//    Iterator itMessages = msgIds.iterator();
//    while (itMessages.hasNext()) {
//      id = (String) itMessages.next();
//      int i = 0;
//      Message message = null;
//      while (i < messages.size()) {
//        message = (Message) messages.get(i);
//        if (id.equals(message.getId())) {
//          messages.remove(i);
//          if ((nbExpirations > 0 ) && message.hasExpiration())
//            nbExpirations -= 1;
//          if (logger.isLoggable(BasicLevel.DEBUG))
//            logger.log(BasicLevel.DEBUG, "Queue.removeMessages msgId = " + id);
//          break;
//        }
//      }
//    }
//  }

  /**
   * get messages, if it's possible.
   * 
   * @param nb
   *            -1 return all messages.
   * @param selector
   *            jms selector.
   * @return List of mom messages.
   */
  private List<Message> getMessages(int nb, String selector, boolean remove) {   
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.getMessages(" + nb + ',' + selector + ',' + remove + ')');

    List<Message> lsMessages = new ArrayList<Message>();
    Message message;
    int j = 0;
    // Checking the deliverable messages:
    while ((lsMessages.size() < nb || nb == -1) &&  j < messages.size()) {
      message = (Message) messages.get(j);

      // If selector matches, sending the message:
      if (Selector.matches(message.getHeaderMessage(), selector) && checkDelivery(message.getHeaderMessage())) {
        message.incDeliveryCount();
        nbMsgsDeliverSinceCreation++;
        // JORAM-326: Saves the queue to avoid bad counter value at restart.
        if (strictCounters) setSave();

        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Queue.getMessages() -> " + j + ',' + message.getId());

        // use in sub class see ClusterQueue
        messageDelivered(message.getId());

        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Message " + message.getId());

        lsMessages.add(message);

        if (remove) {
          messages.remove(message);
          if ((nbExpirations > 0 ) && message.hasExpiration())
            nbExpirations -= 1;
          message.delete();
        } else {
          // message not remove: going on.
          j++;
        }

      } else {
        // If message delivered or selector does not match: going on
        j++;
      }
    }
    return lsMessages;
  }

  private Message getMomMessage(String msgId) {
    Message msg = null;
    for (Iterator<Message> ite = messages.iterator(); ite.hasNext();) {
      msg = ite.next();
      if (msgId.equals(msg.getId()))
        return msg;
    }
    return msg;
  }

  /**
   * Get mom message, delete if remove = true.
   * 
   * @param msgId   message identification
   * @param remove  if true delete message
   * @return mom message
   */
  protected Message getQueueMessage(String msgId, boolean remove) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.getMessage(" + msgId + ',' + remove + ')');

    Message message =  getMomMessage(msgId);
    if (message == null) return null;
    
    // Note (AF): The semantic of this method is unclear, whatever the result of incDeliveryCount the
    // message is returned.
    
    if (checkDelivery(message.getHeaderMessage())) {
    	message.incDeliveryCount();
    	nbMsgsDeliverSinceCreation++;
      // JORAM-326: Saves the queue to avoid bad counter value at restart.
      if (strictCounters) setSave();

      // use in sub class see ClusterQueue
      messageDelivered(message.getId());

    	if (logger.isLoggable(BasicLevel.DEBUG))
    		logger.log(BasicLevel.DEBUG, "Message " + msgId);

    	if (remove) {
    		messages.remove(message);
        if ((nbExpirations > 0 ) && message.hasExpiration())
          nbExpirations -= 1;
    		message.delete();
    	}
    }
    return message;
  }
  
  /**
   * Returns the description of a particular pending message. The message is
   * pointed out through its unique identifier.
   * 
   * @param msgId The unique message's identifier.
   * @return the description of the message.
   * 
   * @see org.objectweb.joram.mom.messages.MessageJMXWrapper
   */
  public CompositeData getMessage(String msgId) throws Exception {
    Message msg = getMomMessage(msgId);
    if (msg == null) return null;
    
    return MessageJMXWrapper.createCompositeDataSupport(msg);
  }

  /**
   * Returns the description of all pending messages.
   * 
   * @return the description of the message.
   * 
   * @see org.objectweb.joram.mom.messages.MessageJMXWrapper
   */
  public TabularData getMessages() throws Exception {
    return MessageJMXWrapper.createTabularDataSupport(messages);
  }
  
  /**
   * Only used  from shell-mom MOMCommandsImpl.
   */
  public List<? extends MessageView> getMessagesView() {
    return messages;
  }
  
  /**
   * Returns the description of a particular delayed message. The message is
   * pointed out through its unique identifier.
   * 
   * @param msgId The unique message's identifier.
   * @return the description of the message.
   * 
   * @see org.objectweb.joram.mom.messages.MessageJMXWrapper
   */
  public CompositeData getDelayedMessage(String msgId) throws Exception {
    Message msg = null;
    for (Iterator<Message> ite = delayed.iterator(); ite.hasNext();) {
      msg = (Message) ite.next();
      if (msgId.equals(msg.getId())) break;
    }
    if (msg == null) return null;

    return MessageJMXWrapper.createCompositeDataSupport(msg);
  }

  /**
   * Returns the description of all pending messages.
   * 
   * @return the description of the message.
   * 
   * @see org.objectweb.joram.mom.messages.MessageJMXWrapper
   */
  public TabularData getDelayedMessages() throws Exception {
    return MessageJMXWrapper.createTabularDataSupport(delayed);
  }

  /**
   * Actually tries to answer the pending "receive" requests.
   * <p>
   * The method may send <code>QueueMsgReply</code> replies to clients.
   *
   * @param index  Index where starting to "browse" the requests.
   */
  protected void deliverMessages(int index) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.deliverMessages(" + index + ')');

    ReceiveRequest notRec = null;
    Message message;
    QueueMsgReply notMsg;
    List<Message> lsMessages = null;

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, " -> requests = " + requests + ')');

    long current = System.currentTimeMillis();
    cleanWaitingRequest(current);
    // Cleaning the possibly expired messages.
    DMQManager dmqManager = cleanPendingMessage(current);

    if (pause) return;
    
    // Processing each request as long as there are deliverable messages:
    while (! messages.isEmpty() && index < requests.size()) {
      notRec = (ReceiveRequest) requests.get(index);
      notMsg = new QueueMsgReply(notRec);

      lsMessages = getMessages(notRec.getMessageCount(), notRec.getSelector(), notRec.getAutoAck());

      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Queue.deliverMessages: notRec.getAutoAck() = " + notRec.getAutoAck() + ", lsMessages = " + lsMessages);

      Iterator<Message> itMessages = lsMessages.iterator();
      while (itMessages.hasNext()) {
        message = (Message) itMessages.next();
        notMsg.addMessage(message.getFullMessage());
        if (!notRec.getAutoAck()) {
          // putting the message in the delivered messages table:
          QueueDelivery queueDelivery = new QueueDelivery(notRec.requester, notRec.getClientContext(), message);
          deliveryTable.put(message.getId(), queueDelivery);
          messages.remove(message);
          if ((nbExpirations > 0 ) && message.hasExpiration())
            nbExpirations -= 1;
        }
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG,
                     "Message " + message.getId() + " to " + notRec.requester + " as reply to " + notRec.getRequestId());
        if (logmsg.isLoggable(BasicLevel.INFO))
          logmsg.log(BasicLevel.INFO,
                     // TODO  (AF): Be careful with backward compatibility, this trace is used for message tracability.
                     // getName() + ": delivers message " + message.getId()  + " to " + notRec.requester + " / " + notRec.getClientContext() + ':' + notRec.getRequestId());
                     getName() + ": delivers message " + message.getId()  + " to " + notRec.requester + " / " + notRec.getRequestId());
      }

      if (isLocal(notRec.requester)) {
        notMsg.setPersistent(false);
      }

      // The deliveryTable is saved outside the Queue agent.
      /*
      if (notMsg.isPersistent() && !notRec.getAutoAck()) {
        // state change, so save.
        setSave();
      }*/

      // Next request:
      if (notMsg.getSize() > 0) {
        requests.remove(index);
        forward(notRec.requester, notMsg);
      } else {
        index++;
      }
    }
    // If needed, sending the dead messages to the DMQ:
    if (dmqManager != null)
      dmqManager.sendToDMQ();
  }

  /**
   * Returns true if conditions are ok to deliver the message.
   * This method must be overloaded in subclasses.
   * Be careful only the message header is accessible.
   */
  protected boolean checkDelivery(org.objectweb.joram.shared.messages.Message msg) {
    return true;
  }

  /** 
   * call in deliverMessages just after forward(msg),
   * overload this method to process a specific treatment.
   */
  protected void messageDelivered(String msgId) {}

  /** 
   * call in deliverMessages just after a remove message (invalid),
   * overload this method to process a specific treatment.
   */
  protected void messageRemoved(String msgId) {}

  /**
   * Returns <code>true</code> if a given message is considered as  undeliverable,
   * because its delivery count matches the queue's threshold, if any, or the
   * server's default threshold value (if any).
   */
  protected boolean isUndeliverable(Message message) {
    // By default, the queue threshold is -1 and we have to use the global threshold.
    // If the queue threshold is 0, the user has explicitly disabled the mechanism for this queue.
    if (threshold == 0) return false;
    if (threshold > 0)
      return (message.getDeliveryCount() >= threshold);
    else if (Queue.getDefaultThreshold() > 0)
      return (message.getDeliveryCount() >= Queue.getDefaultThreshold());
    return false;
  }

  /**
   * Adds the client messages in the queue.
   * This method is used when messages does not come from a JMS Client, for example
   * for AcquisitionQueue, AliasQueue and AliasInQueue.
   * 
   * @param clientMsgs client message notification.
   * @param throwsExceptionOnFullDest true, can throws an exception on sending message on full destination
   * @throws AccessException
   */
  public void addClientMessages(ClientMessages clientMsgs, boolean throwsExceptionOnFullDest) throws AccessException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Queue.addClientMessage(" + clientMsgs + ')');
    
    long currentTime = System.currentTimeMillis();
    long deliveryTime = currentTime;
    if (deliveryDelay > 0)
      deliveryTime += deliveryDelay;

    if (clientMsgs != null) {
      Message msg;
      // Storing each received message:
      for (Iterator<org.objectweb.joram.shared.messages.Message> msgs = clientMsgs.getMessages().iterator(); msgs.hasNext();) {
        msg = new Message(msgs.next());
        msg.order = arrivalState.getAndIncrementArrivalCount(msg.isPersistent());
        
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Queue.addClientMessage() -> " + msg.getId() + ',' + msg.order);

        if ((deliveryTime > currentTime) && (deliveryTime > msg.getDeliveryTime()))
          msg.setDeliveryTime(deliveryTime);;

        if (interceptorsAvailable()) {
        	// Gets the shared message (useless, we have just created this message)
        	org.objectweb.joram.shared.messages.Message message = msg.getFullMessage();
        	// set the destination name
        	message.setProperty("JoramDestinationName", getName());
        	// interceptors process
      		org.objectweb.joram.shared.messages.Message m = processInterceptors(message);
      		if (m == null) {
      			// The message is rejected, sends it to the DMQ.
      			DMQManager dmqManager = new DMQManager(dmqId, getId());
            nbMsgsSentToDMQSinceCreation++;
            // JORAM-326: Saves the queue to avoid bad counter value at restart.
            if (strictCounters) setSave();
            dmqManager.addDeadMessage(msg.getFullMessage(), MessageErrorConstants.INTERCEPTORS);
            dmqManager.sendToDMQ();
      			continue;
      		} else {
      		  // Builds a new MOM message with modified 'shared message'.
      			msg = new org.objectweb.joram.mom.messages.Message(m);
      		}
      	}
        
        if (msg.getDeliveryTime() > currentTime) {
          // TODO: We can not set the client context id, fix to -1. Is it a problem?
          addDeliveryTimeMessage(msg, throwsExceptionOnFullDest, false);
        } else {
          // store message
          storeMessage(msg, throwsExceptionOnFullDest);
        }
      }
    }
    // Launching a delivery sequence:
    deliverMessages(0);
  }

  protected void handleExpiredNot(AgentId from, ExpiredNot not) {
    Notification expiredNot = not.getExpiredNot();
    List<org.objectweb.joram.shared.messages.Message> messagesToHandle;
    // ClientMessages and TopicMsgsReply are the notifications which can expire in the networks.
    // QueueMsgReply can't expire due to protocol limitations
    if (expiredNot instanceof ClientMessages) {
      messagesToHandle = ((ClientMessages) expiredNot).getMessages();
    } else if (expiredNot instanceof TopicMsgsReply) {
      messagesToHandle = ((TopicMsgsReply) expiredNot).getMessages();
    } else {
      if (logger.isLoggable(BasicLevel.ERROR))
        logger.log(BasicLevel.ERROR,
                   "Expired notification holds an unknown notification: " + expiredNot.getClass().getName());
      return;
    }

    // Let senderId to null because we want to explicitly send messages to the queue itself.
    DMQManager dmqManager = new DMQManager(getId(), null);
    Iterator<org.objectweb.joram.shared.messages.Message> iterator = messagesToHandle.iterator();
    while (iterator.hasNext()) {
      dmqManager.addDeadMessage((org.objectweb.joram.shared.messages.Message) iterator.next(),
                                MessageErrorConstants.EXPIRED);
    }
    dmqManager.sendToDMQ();
  }

	// Get flow Control related informations.
	// TODO AF: may be we can use generic Destination.getJMXStatistics method.
	protected fr.dyade.aaa.common.stream.Properties getStats() {
	  fr.dyade.aaa.common.stream.Properties stats = new fr.dyade.aaa.common.stream.Properties();
	  
    // TODO (AF): Should be getNbMsgsDeliverSinceCreation?
		stats.put("NbMsgsDeliverSinceCreation", nbMsgsDeliverSinceCreation);
		stats.put("PendingMessageCount", getPendingMessageCount());
		
		return stats;
	}
	
	// ================================================================================
	// Handling of AdminSyncNotification
	
  /**
   * Exports all messages in the specified directory.
   * Options? filtrage? Format?
   * @param dirpath
   * @param selector
   * @param quiet
   * @throws Exception 
   * @throws InterruptedException 
   */
  public void exportMessages(String dirpath, String selector, boolean binary) throws Exception {
    AdminSyncNotification cmd = new AdminSyncNotification(AdminSyncNotification.EXPORT, true);
    cmd.addParameter(AdminSyncNotification.EXPORT_DIRPATH, dirpath);
    cmd.addParameter(AdminSyncNotification.EXPORT_SELECTOR, selector);
    cmd.addParameter(AdminSyncNotification.EXPORT_BINARY, binary?"true":"false");
    cmd.invoke(getId());
    return;
  }
  
  /**
   * Handling of AdminSyncNotification.
   * Currently only the EXPORT command is handled.
   * 
   * @param from
   * @param not
   */
  protected void doAdminSyncNotification(AgentId from, AdminSyncNotification not) {
    int command = not.getCommand();
    try {
      if (command == AdminSyncNotification.EXPORT) {
        String dirpath = not.getParameter(AdminSyncNotification.EXPORT_DIRPATH);
        String selector = not.getParameter(AdminSyncNotification.EXPORT_SELECTOR);
        Boolean binary = Boolean.valueOf(not.getParameter(AdminSyncNotification.EXPORT_BINARY));
        doExportMessages(dirpath, selector, binary);
        not.Return(new Object[] {"Unkonw command: " + command});
      } else if (command == AdminSyncNotification.CLEAN_PENDING_MESSAGE) {
        if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG, this.toString() + ".doAdminSyncNotification CLEAN_PENDING_MESSAGE");

        cleanPendingMessage(System.currentTimeMillis());
        not.Return(null);
      } else if (command == AdminSyncNotification.CLEAN_WAITING_REQUEST) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.WARN, this.toString() + ".doAdminSyncNotification CLEAN_WAITING_REQUEST");

        cleanWaitingRequest(System.currentTimeMillis());
        not.Return(null);
      } else {
        super.doAdminSyncNotification(from, not);
        not.Return(null);
      }
    } catch (Exception exc) {
      logger.log(BasicLevel.WARN,
                 this.toString() + ".doAdminSyncNotification, error executing command " + command + ": ", exc);
      not.Throw(exc);
    }
  }

  /**
   * Exports all messages in the specified directory.
   * Be careful, this method needs to be called from an agent reaction.
   * @throws IOException 
   */
  private void doExportMessages(String dirpath, String selector, boolean binary) throws IOException {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 this.toString() + ".doExportMessages, export mesasges to " + dirpath);
    
    // Creates the directory, then export each message in a separate file using JSon format.
    // TODO (AF): May be we have to create a specific sub-directory for the queue.
    File dir = new File(dirpath);
    if (!dir.mkdirs() && (!dir.exists() || !dir.isDirectory()))
      throw new IOException("Cannot create directory " + dir.getPath());
    
    Message message;
    // Gets all ready to deliver messages.
    for (int i = 0; i < messages.size(); i++) {
      message = (Message) messages.get(i);
      if (Selector.matches(message.getHeaderMessage(), selector)) {
        // Create a file for the message then writes it in JSon format.
        try {
          message.exportJSonToFile(dir, binary);
        } catch (IOException exc) {
          logger.log(BasicLevel.WARN,
                     this.toString() + ".doExportMessages, cannot export " + message.getId() + ": " + exc);
        }
      }
    }
    // Get delayed messages
    for (int i = 0; i < delayed.size(); i++) {
      message = (Message) delayed.get(i);
      if (Selector.matches(message.getHeaderMessage(), selector)) {
        // Create a file for the message then writes it in JSon format.
        try {
          message.exportJSonToFile(dir, binary);
        } catch (IOException exc) {
          logger.log(BasicLevel.WARN,
                     this.toString() + ".doExportMessages, cannot export " + message.getId() + ": " + exc);
        }
      }
    }
  }
  
	// Encodable interface
	
	public int getEncodableClassId() {
    return JoramHelper.QUEUE_CLASS_ID;
  }
  
  public int getEncodedSize() throws Exception {
    int encodedSize = super.getEncodedSize();
    encodedSize += INT_ENCODED_SIZE * 6;
    encodedSize += BOOLEAN_ENCODED_SIZE;
    encodedSize += LONG_ENCODED_SIZE;
    for (ReceiveRequest request : requests) {
      encodedSize += request.getEncodedSize();
    }
    return encodedSize;
  }
  
  public void encode(Encoder encoder) throws Exception {
    super.encode(encoder);
    encoder.encodeUnsignedInt(nbMaxMsg);
    encoder.encodeUnsignedLong(nbMsgsDeniedSinceCreation);
    encoder.encodeUnsignedInt(priority);
    encoder.encodeSignedInt(threshold);
    encoder.encodeBoolean(pause);
    encoder.encodeSignedInt(deliveryDelay);
    encoder.encodeSignedInt(redeliveryDelay);
    encoder.encodeUnsignedInt(requests.size());
    for (ReceiveRequest request : requests) {
      request.encode(encoder);
    }
  }

  public void decode(Decoder decoder) throws Exception {
    super.decode(decoder);
    nbMaxMsg = decoder.decodeUnsignedInt();
    nbMsgsDeniedSinceCreation = decoder.decodeUnsignedLong();
    priority = decoder.decodeUnsignedInt();
    threshold = decoder.decodeSignedInt();
    pause = decoder.decodeBoolean();
    deliveryDelay = decoder.decodeSignedInt();
    redeliveryDelay = decoder.decodeSignedInt();
    int requestsSize = decoder.decodeUnsignedInt();
    requests = new Vector<ReceiveRequest>(requestsSize);
    for (int i = 0; i < requestsSize; i++) {
      ReceiveRequest request = new ReceiveRequest();
      request.decode(decoder);
      requests.add(request);
    }
  }
  
  public static class QueueFactory implements EncodableFactory {

    public Encodable createEncodable() {
      // Use this specific constructor to avoid the allocation of a new stamp by the Agent
      // constructor. We should create a specific Agent constructor to use with the Encodable
      // framework.
      return new Queue(null, false, AgentId.MinWKSIdStamp);
    }
    
  }
}
