/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2026 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.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.Vector;

import org.objectweb.joram.mom.dest.AdminTopic.DestinationDesc;
import org.objectweb.joram.mom.notifications.AbstractRequestNot;
import org.objectweb.joram.mom.notifications.ClientMessages;
import org.objectweb.joram.mom.notifications.ExceptionReply;
import org.objectweb.joram.mom.notifications.FwdAdminRequestNot;
import org.objectweb.joram.mom.notifications.GetRightsReplyNot;
import org.objectweb.joram.mom.notifications.GetRightsRequestNot;
import org.objectweb.joram.mom.notifications.PingNot;
import org.objectweb.joram.mom.notifications.PongNot;
import org.objectweb.joram.mom.notifications.RequestGroupNot;
import org.objectweb.joram.mom.notifications.WakeUpNot;
import org.objectweb.joram.mom.proxies.SendRepliesNot;
import org.objectweb.joram.mom.proxies.SendReplyNot;
import org.objectweb.joram.mom.util.DMQManager;
import org.objectweb.joram.mom.util.InterceptorsHelper;
import org.objectweb.joram.mom.util.MessageInterceptor;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.MessageErrorConstants;
import org.objectweb.joram.shared.admin.AdminCommandConstant;
import org.objectweb.joram.shared.admin.AdminCommandReply;
import org.objectweb.joram.shared.admin.AdminCommandRequest;
import org.objectweb.joram.shared.admin.AdminReply;
import org.objectweb.joram.shared.admin.AdminRequest;
import org.objectweb.joram.shared.admin.DeleteDestination;
import org.objectweb.joram.shared.admin.GetStatsReply;
import org.objectweb.joram.shared.admin.GetStatsRequest;
import org.objectweb.joram.shared.admin.SetDMQRequest;
import org.objectweb.joram.shared.admin.SetReader;
import org.objectweb.joram.shared.admin.SetRight;
import org.objectweb.joram.shared.admin.SetWriter;
import org.objectweb.joram.shared.admin.UnsetReader;
import org.objectweb.joram.shared.admin.UnsetWriter;
import org.objectweb.joram.shared.excepts.AccessException;
import org.objectweb.joram.shared.excepts.MomException;
import org.objectweb.joram.shared.excepts.RequestException;
import org.objectweb.joram.shared.messages.ConversionHelper;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.joram.shared.messages.MessageHelper;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.Agent;
import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.agent.CallbackNotification;
import fr.dyade.aaa.agent.Channel;
import fr.dyade.aaa.agent.DeleteNot;
import fr.dyade.aaa.agent.Notification;
import fr.dyade.aaa.agent.UnknownAgent;
import fr.dyade.aaa.agent.UnknownNotificationException;
import fr.dyade.aaa.agent.WakeUpTask;
import fr.dyade.aaa.common.Debug;
import fr.dyade.aaa.common.encoding.Decoder;
import fr.dyade.aaa.common.encoding.EncodableHelper;
import fr.dyade.aaa.common.encoding.Encoder;
import fr.dyade.aaa.common.stream.StreamUtil;
import fr.dyade.aaa.util.management.MXWrapper;

/**
 * The <code>Destination</code> class implements the common behavior of
 * MOM destinations.
 */
public abstract class Destination extends Agent implements DestinationMBean {
  /**
   * define serialVersionUID for interoperability, fix with 5.15.0 value.
   * This value is identical in 5.15.0, 5.16.0, 5.16.3, 5.17 and next. It is different
   * in 5.16.1 and 5.16.2 (and 5.17.0RC).
   */
  private static final long serialVersionUID = -4457639325610333885L;
  
  private static final Logger logger = Debug.getLogger(Destination.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);
  
  // JORAM-323: Verify if the destination is temporary.
  protected transient boolean temporary = false;
  
  /**
   * <code>true</code> if the destination successfully processed a deletion
   * request.
   */
  private boolean deletable = false;

  /** period to run task at regular interval: cleaning, load-balancing, etc. */
  private long period = -1L;
  
  /**
   * Identifier of the destination's administrator.
   * In any case the administration topics are authorized to handle the
   * destination, this mechanism allows an other agent to get the same rights.
   * In particular it is needed to allow user's proxy to handle temporary
   * destinations.
   */
  private AgentId adminId;
  
  protected transient WakeUpTask task;

  /** the interceptors list. */
  private List<Properties> interceptorsProp = null;
  private transient List<MessageInterceptor> interceptors = null;
  
  /**
   * Empty constructor for newInstance().
   */
  public Destination() {
    super();
  }

  /**
   * Constructor with parameters for fixing the destination and specifying its identifier.
   * Be careful, this constructor should only be used for destinations with well known stamps, it
   * throws an exception otherwise.
   * 
   * @param name  Name of destination;
   * @param fixed If true, destination is fixed in memory.
   * @param stamp Specific stamp for resulting agent.
   */
  protected Destination(String name, boolean fixed, int stamp) {
    super(name, fixed, stamp);
  }

  /**
   * Returns the type of this destination: Queue or Topic.
   * 
   * @return the type of this destination.
   * @see org.objectweb.joram.shared.DestinationConstants#TOPIC_TYPE
   * @see org.objectweb.joram.shared.DestinationConstants#QUEUE_TYPE
   * @see org.objectweb.joram.shared.DestinationConstants#TEMPORARY
   */
  public abstract byte getType();

  /**
   * Gives this agent an opportunity to initialize after having been deployed,
   * and each time it is loaded into memory.
   * 
   * @param firstTime true when first called by the factory
   * @exception Exception
   *              unspecialized exception
   */
  protected void agentInitialize(boolean firstTime) throws Exception {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, getName() + ".agentInitialize(" + firstTime + ')');

    super.agentInitialize(firstTime);

    // JORAM-323: Verify if the destination is temporary.
    if ((getId().getStamp() != AgentId.JoramAdminStamp) &&
        // JMQ-142:  May be we should just verify that the agent is a special one (id < MaxWKSIdStamp).
        (getId().getStamp() != AgentId.MqttDefaultTopicStamp) &&
        (getId().getStamp() != AgentId.MqttDefault$TopicStamp)) {
      DestinationDesc desc = AdminTopic.lookupDest(getName(), getType());
      if (desc == null) {
        logger.log(BasicLevel.WARN,  "Cannot find destination in registry:" + getName());
      } else {
        if (DestinationConstants.isTemporary(desc.getType())) {
          if (DEBUG)
            logger.log(BasicLevel.DEBUG, "Set temporary field, destination will be deleted:" + getName());
          temporary = true;
        }
      }
    }

    // interceptors
    if (interceptorsProp != null) {
      interceptors = new ArrayList<MessageInterceptor>();
      InterceptorsHelper.addInterceptors(getDestinationId(), getName(), AdminCommandConstant.INTERCEPTORS, interceptorsProp, interceptors);
    }

    initialize(firstTime);

    if (getPeriod() > -1)
      task = new WakeUpTask(getId(), WakeUpNot.class, getPeriod());

    try {
      MXWrapper.registerMBean(this, getMBeanName());
    } catch (Exception exc) {
      logger.log(BasicLevel.ERROR, this + " jmx failed", exc);
    }
  }

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

  /**
   * Finalizes the agent before it is garbaged.
   * 
   * @param last true if the destination is deleted
   */
  public void agentFinalize(boolean last) {
    if (task != null) task.cancel();
    
    finalize(last);
    
    try {
      MXWrapper.unregisterMBean(getMBeanName());
    } catch (Exception exc) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "Destination.agentFinalize", exc);
    }
    super.agentFinalize(last);
  }

  /**
   * Finalizes the destination before it is garbaged.
   * 
   * @param last true if the destination is deleted
   */
  protected abstract void finalize(boolean last);

  private final String getMBeanName() {
    StringBuffer strbuf = new StringBuffer();

    strbuf.append("Joram#").append(AgentServer.getServerId());
    strbuf.append(':');
    strbuf.append("type=Destination,name=").append(getName());

    return strbuf.toString();
  }

  abstract fr.dyade.aaa.common.stream.Properties getStats();
  
  /**
   * Distributes the received notifications to the appropriate reactions.
   * 
   * @throws Exception
   */
  public void react(AgentId from, Notification not) throws Exception {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "Destination.react(" + from + ',' + not + ')');

    // set agent no save (this is the default).
    setNoSave();
    
    CallbackNotification callbackNotification = null;

    try {
      if (not instanceof GetRightsRequestNot)
        getRights(from, (GetRightsRequestNot) not);
      else if (not instanceof ClientMessages) {
        ClientMessages clientMessages = (ClientMessages) not;
        callbackNotification = clientMessages;
        clientMessages(from, clientMessages);
      } else if (not instanceof UnknownAgent)
        unknownAgent(from, (UnknownAgent) not);
      else if (not instanceof RequestGroupNot)
        requestGroupNot(from, (RequestGroupNot) not);
      else if (not instanceof DeleteNot) {
        deleteNot(from, (DeleteNot) not);
        if (canBeDeleted()) {
          // A DeleteNot notification is finally processed at the
          // Agent level when its processing went successful in
          // the DestinationItf instance.
          super.react(from, not);
        }
      } else if (not instanceof WakeUpNot) {
        if (DEBUG) {
          logger.log(BasicLevel.DEBUG,
                     "wakeupnot received: current task=" + task + " update=" + ((WakeUpNot) not).isUpdate());
        }
        setNoSave();
        if (task == null || ((WakeUpNot) not).isUpdate()) {
          doSetPeriod(getPeriod());
        }
        if (getPeriod() > 0) {
          wakeUpNot((WakeUpNot) not);
        }
      } else if (not instanceof FwdAdminRequestNot) {
        handleAdminRequestNot(from, (FwdAdminRequestNot) not);
      } else if (not instanceof PingNot) {
        Channel.sendTo(from, new PongNot(getStats()));
      } else {
        throw new UnknownNotificationException(not.getClass().getName());
      }
    } catch (MomException exc) {
      // MOM Exceptions are sent to the requester.
      if (DEBUG)
        logger.log(BasicLevel.WARN, getName() + ".react()", exc);
      else
        logger.log(BasicLevel.WARN, getName() + ".react(): " + exc.getMessage());
      if (callbackNotification != null) {
        callbackNotification.failed(exc); 
      } 
      AbstractRequestNot req = (AbstractRequestNot) not;
      Channel.sendTo(from, new ExceptionReply(req, exc));
    } catch (UnknownNotificationException exc) {
      super.react(from, not);
    }
  }

  private void doSetPeriod(long period) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, this + ": setPeriod(" + period + ")." + " -> task " + task);
    if (task == null) {
      task = new WakeUpTask(getId(), WakeUpNot.class, period);
    } else {
      // cancel task
      task.cancel();
      // Schedules the wake up task period.
      if (period > 0)
        task = new WakeUpTask(getId(), WakeUpNot.class, period);
    }
  }
  
  /** <code>true</code> if the READ access is granted to everybody. */
  protected boolean freeReading = false;
  /** <code>true</code> if the WRITE access is granted to everybody. */
  protected boolean freeWriting = false;
  /** Table of the destination readers and writers. */
  protected Hashtable<AgentId, Integer> clients = new Hashtable<AgentId, Integer>();

  /** READ access value. */
  public static final Integer READ = Integer.valueOf(1);
  /** WRITE access value. */
  public static final Integer WRITE = Integer.valueOf(2);
  /** READ and WRITE access value. */
  public static final Integer READWRITE = Integer.valueOf(3);

  /**
   * Identifier of the dead message queue this destination must send its
   * dead messages to, if any.
   */
  protected AgentId dmqId = null;

  /**
   * Transient <code>StringBuffer</code> used to build message, this buffer
   * is created during agent initialization, then reused during the destination
   * life.
   */
  transient StringBuffer strbuf = new StringBuffer();

  /**
   * date of creation.
   */
  private long creationDate = System.currentTimeMillis();
  
  // JORAM-326: 
  static protected boolean strictCounters;
  
  static {
    strictCounters = AgentServer.getBoolean(DestinationConstants.DEST_STRICT_CPT);
    // Be careful, may be logger is not already initialized!
    Debug.getLogger(Destination.class.getName()).log(BasicLevel.INFO, "Force strict values for queue counters: " + strictCounters);
  }

  protected long nbMsgsDeliverSinceCreation = 0;
  protected long nbMsgsSentToDMQSinceCreation = 0;

  /**
   * Sets the administrator of the destination.
   * 
   * @param adminId Identifier of the administrator of the destination.
   */
  public final void setAdminId(AgentId adminId) {
    this.adminId = adminId;
  }

  /**
   * Sets the configuration of a <code>Destination</code>. Be careful, this is
   * done a first time before {@link #deploy()}, so the agent is serialized and
   * initialized afterwards.<br>
   * After deployment, firstTime argument is set to false.
   * 
   * @param prop The initial set of properties.
   */ 
  public void setProperties(Properties prop, boolean firstTime) throws Exception {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, this + ", setProperties.");

    long newPeriod = -1;
    if (prop != null && prop.containsKey(DestinationConstants.WAKEUP_PERIOD)) {
      try {
        newPeriod = ConversionHelper.toLong(prop.get(DestinationConstants.WAKEUP_PERIOD));
      } catch (Exception e) {
        logger.log(BasicLevel.ERROR, this + ": error setting destination period", e);
      }
    }
    
    if (firstTime) {
      period = newPeriod;
      if (prop != null && prop.containsKey(DestinationConstants.FIXED_IN_MEMORY))
        fixed = ConversionHelper.toBoolean(prop.get(DestinationConstants.FIXED_IN_MEMORY));
    } else {
      setPeriod(newPeriod);
      if (prop != null && prop.containsKey(DestinationConstants.FIXED_IN_MEMORY)) {
        boolean fixedInMemory = ConversionHelper.toBoolean(prop.get(DestinationConstants.FIXED_IN_MEMORY));
        if (fixed != fixedInMemory)
          logger.log(BasicLevel.WARN, this + ": cannot change fixed attribute of an already created destination.");

      }
    }
    
    interceptorsProp = null;
    interceptors = null;
    if (prop != null && prop.containsKey(AdminCommandConstant.INTERCEPTORS)) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, this + ": setProperties interceptors = " + prop.get(AdminCommandConstant.INTERCEPTORS));
      addInterceptor(getDestinationId(), getName(), prop);
    }
  }

  protected boolean isLocal(AgentId id) {
    return (getId().getTo() == id.getTo());
  }

  /** Returns <code>true</code> if the destination might be deleted. */
  private final boolean canBeDeleted() {
    return deletable;
  }

  /**
   * Returns  the period value of this destination, -1 if not set.
   *
   * @return the period value of this destination; -1 if not set.
   */
  public long getPeriod() {
    return period;
  }

  /**
   * Sets or unsets the period for this destination.
   *
   * @param period The period value to be set or -1 for unsetting previous
   *               value (ignore 0).
   */
  public void setPeriod(long period) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "setPeriod " + period + ", old was " + this.period);
    
    if (this.period != period) {
      WakeUpNot not = new WakeUpNot(true);
      forward(getId(), not);
      this.period = period;
    }
  }
  
  public abstract void wakeUpNot(WakeUpNot not);
    
  /**
   * Method implementing the reaction to a <code>SetRightRequest</code>
   * notification requesting rights to be set for a user.
   * 
   * @param request       ACL
   * @param replyTo       Identifier of the sender
   * @param requestMsgId  identifier of the request
   * @param replyMsgId    identifier of the reply
   */
  protected void setRight(SetRight request,
                       AgentId replyTo,
                       String requestMsgId,
                       String replyMsgId) {
    AgentId user = null;
    if (request.getUserProxId() != null)
      user = AgentId.fromString(request.getUserProxId());

    int right = 0;
    if (request instanceof SetReader)
      right = READ;
    else if (request instanceof SetWriter)
      right = WRITE;
    else if (request instanceof UnsetReader)
      right = - READ;
    else if (request instanceof UnsetWriter)
      right = - WRITE;

    try {
      processSetRight(user, right);
      doRightRequest(user, right);

      replyToTopic(new AdminReply(true, null), replyTo, requestMsgId, replyMsgId);
    } catch (RequestException exc) {
      strbuf.append("Request [").append(request.getClass().getName());
      strbuf.append("], sent to Destination [").append(getId());
      strbuf.append("], successful [false]: ").append(exc.getMessage());
      replyToTopic(new AdminReply(false, strbuf.toString()), replyTo, requestMsgId, replyMsgId);
      logger.log(BasicLevel.ERROR, strbuf.toString());
      strbuf.setLength(0);
    }
  }
 
  /**
   * This method is needed for right revocation.
   * It allows to remove request or subscription from users no longer authorized.
   * 
   * @param user  The user about right modification.
   * @param right The right modification.
   */
  abstract protected void doRightRequest(AgentId user, int right);

  /** set user right. */
  protected void processSetRight(AgentId user, int right) 
    throws RequestException {
    // state change, so save.
    setSave();

    // Setting "all" users rights:
    if (user == null) {
      if (right == READ)
        freeReading = true;
      else if (right == WRITE)
        freeWriting = true;
      else if (right == -READ)
        freeReading = false;
      else if (right == -WRITE)
        freeWriting = false;
      else
        throw new RequestException("Invalid right value: " + right);
    } else {
      // Setting a specific user right:
      Integer currentRight = (Integer) clients.get(user);
      if (right == READ) {
        if (currentRight != null && currentRight.intValue() == WRITE)
          clients.put(user, READWRITE);
        else
          clients.put(user, READ);
      } else if (right == WRITE) {
        if (currentRight != null && currentRight.intValue() == READ)
          clients.put(user, READWRITE);
        else
          clients.put(user, WRITE);
      } else if (right == -READ) {
        if (currentRight != null && currentRight.intValue() == READWRITE)
          clients.put(user, WRITE);
        else if (currentRight != null && currentRight.intValue() == READ)
          clients.remove(user);
      } else if (right == -WRITE) {
        if (currentRight != null && currentRight.intValue() == READWRITE)
          clients.put(user, READ);
        else if (currentRight != null && currentRight.intValue() == WRITE)
          clients.remove(user);
      } else
        throw new RequestException("Invalid right value: " + right);
    }
  }

  /**
   * Method implementing the reaction to a <code>GetRightsRequest</code>
   * notification requesting the rights about this destination.
   *
   * @exception AccessException  If the requester is not the administrator.
   */
  protected void getRights(AgentId from, GetRightsRequestNot not) throws AccessException {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    AgentId key;
    int right;
    Vector<AgentId> readers = new Vector<AgentId>();
    Vector<AgentId> writers = new Vector<AgentId>();
    for (Enumeration<AgentId> keys = clients.keys(); keys.hasMoreElements();) {
      key = keys.nextElement();
      right = clients.get(key).intValue();

      if (right == READ || right == READWRITE) readers.add(key);
      if (right == WRITE || right == READWRITE) writers.add(key);
    }
    forward(from, new GetRightsReplyNot(not, freeReading, freeWriting, readers, writers));
  }


  private static String[] _rights = {":R;", ";W;", ":RW;"};

  /**
   * Returns a string representation of the rights set on this destination.
   *
   * @return the rights set on this destination.
   */
  public String[] getRights() {
    String rigths[] = new String[clients.size()];

    AgentId key;
    int right;

    int i=0;
    for (Enumeration<AgentId> keys = clients.keys(); keys.hasMoreElements();) {
      key = keys.nextElement();
      right = clients.get(key).intValue();
      rigths[i++] = key.toString() + _rights[right -1];
    }

    return rigths;
  }

  /**
   * Returns a string representation of rights set on this destination for a
   * particular user. The user is pointed out by its unique identifier.
   *
   * @param userid The user's unique identifier.
   * @return the rights set on this destination.
   */
  public String getRight(String userid) {
    AgentId key = AgentId.fromString(userid);
    if (key == null) return userid + ":bad user;";
    Integer right = (Integer) clients.get(key);
    if (right == null) return userid + ":unknown;";

    return userid + _rights[right.intValue() -1];
  }

  /**
   * This method allows to exclude some JMX attribute of getJMXStatistics method.
   * It must be overloaded in subclass.
   * 
   * @param attrName name of attribute to test.
   * @return true if the attribute is a valid one.
   */
  protected boolean isValidJMXAttribute(String attrName) {
    if (attrName == null)
      return false;
    return true;
  }
  
  /**
   * Returns values of all valid JMX attributes about the destination.
   * 
   * @param names A comma separated list of requested JMX attribute names.
   * @return a Hashtable containing the values of all valid JMX attributes about the destination.
   *         The keys are the name of corresponding attributes.
   */
  protected final Hashtable<String, Object> getJMXStatistics(String names) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "Destination.getJMXStatistics()");

    Hashtable<String, Object> stats = null;

    try {
      if (names != null) {
        StringTokenizer st = new StringTokenizer(names, ",");
        stats = new Hashtable<String, Object>(st.countTokens());
        while (st.hasMoreTokens()) {
          String name = st.nextToken();
          if (isValidJMXAttribute(name)) {
            Object value = MXWrapper.getAttribute(getMBeanName(), name);
            if ((value != null) && StreamUtil.isStreamable(value))
              stats.put(name, value);
          }

        }
      } else {
        List<String> attributes = MXWrapper.getAttributeNames(getMBeanName());
        if (attributes != null) {
          stats = new Hashtable<String, Object>(attributes.size());
          for (int k = 0; k < attributes.size(); k++) {
            String name = attributes.get(k);
            if (isValidJMXAttribute(name)) {
              Object value = MXWrapper.getAttribute(getMBeanName(), name);
              if ((value != null) && StreamUtil.isStreamable(value))
                stats.put(name, value);
            }
          }
        }
      }
    } catch (Exception exc) {
      logger.log(BasicLevel.ERROR, " getAttributes  on " + getMBeanName() + " error.", exc);
    }

    return stats;
  }

  /**
   * Method implementing the reaction to a <code>ClientMessages</code>
   * notification holding messages sent by a client.
   * <p>
   * If the sender is not a writer on the destination the messages are
   * sent to the DMQ and an exception is thrown. Otherwise, the processing of
   * the received messages is performed in subclasses.
   *
   * @exception AccessException  If the sender is not a WRITER on the
   *              destination.
   */
  protected void clientMessages(AgentId from, ClientMessages not) throws AccessException {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "Destination.clientMessages(" + from + ',' + not + ')');
    
    if (AgentId.nullId.equals(from)) {
      from = not.getProxyId();
    }
    
    // If sender is not a writer, sending the messages to the DMQ, and
    // throwing an exception:
    if (!isWriter(from)) {
      DMQManager dmqManager = new DMQManager(not.getDMQId(), dmqId, getId());
      Message msg;
      for (Iterator<Message> msgs = not.getMessages().iterator(); msgs.hasNext();) {
        msg = msgs.next();
        nbMsgsSentToDMQSinceCreation++;
        // JORAM-326: Saves the queue to avoid bad counter value at restart.
        if (strictCounters) {
          setSave();
        }
        
        dmqManager.addDeadMessage(msg, MessageErrorConstants.NOT_WRITEABLE);
        handleDeniedMessage(msg.id, AgentId.fromString(msg.replyToId));
      }
      dmqManager.sendToDMQ();
      throw new AccessException("WRITE right not granted");
    }

    // TODO: remove 'from'
    doClientMessages(from, not, true);

    // For topic performance we must send reply after process ClientMessage. It results
    // in a best flow-control of sender allowing the handling of forwarded messages before
    // sender freeing.
    
    if (!not.isPersistent() && !not.getAsyncSend() && !not.hasCallback()) {
      forward(from, new SendReplyNot(not.getClientContext(), not.getRequestId()));
    }
  }

  /**
   * Method used to do specific actions when a message is denied because of a
   * lack of rights.
   */
  protected void handleDeniedMessage(String msgId, AgentId replyTo) {
    // Nothing to do, useful in admin topic
  }

  /**
   * Method implementing the reaction to an <code>UnknownAgent</code>
   * notification.
   * <p>
   * If the unknown agent is the DMQ, its identifier is set to null. If it
   * is a client of the destination, it is removed. Specific processing is
   * also done in subclasses.
   */
  protected void unknownAgent(AgentId from, UnknownAgent not) {
    if (isAdministrator(not.agent)) {
      if (logger.isLoggable(BasicLevel.ERROR))
            logger.log(BasicLevel.ERROR,
                       "Admin of dest " + getId() + " does not exist anymore.");
    } else if (not.agent.equals(dmqId)) {
      // state change, so save.
      setSave();
      dmqId = null;
    } else {
      // state change, so save.
      setSave();
      clients.remove(from);
      doUnknownAgent(not);
    }
  }

  /**
   * Method implementing the reaction to a <code>DeleteNot</code>
   * notification requesting the deletion of the destination.
   * <p>
   * The processing is done in subclasses if the sender is an administrator.
   */
  protected void deleteNot(AgentId from, DeleteNot not) {
    if (! isAdministrator(from)) {
      if (logger.isLoggable(BasicLevel.WARN))
        logger.log(BasicLevel.WARN,
                   "Unauthorized deletion request from " + from);
    } else {
      // JORAM-323: Removes the destination from AdminTopic registry.
      AdminTopic.unregisterDest(getName());
      doDeleteNot(not);
      setSave(); // state change, so save.
      deletable = true;
    }
  }
  
  protected void requestGroupNot(AgentId from, RequestGroupNot not) {
    Enumeration<ClientMessages> en = not.getClientMessages();
    ClientMessages theCM = en.nextElement();
    Vector<SendReplyNot> replies = new Vector<SendReplyNot>();
    replies.addElement(new SendReplyNot(theCM.getClientContext(), theCM.getRequestId()));
    while (en.hasMoreElements()) {
      ClientMessages cm = en.nextElement();
      List<Message> msgs = cm.getMessages();
      for (int i = 0; i < msgs.size(); i++) {
        theCM.addMessage(msgs.get(i));
      }
      if (! cm.getAsyncSend()) {
        replies.addElement(new SendReplyNot(cm.getClientContext(), cm.getRequestId()));
      }
    }
    
    
    try {
      doClientMessages(from, theCM, false);
    } catch (AccessException e) {/* never happens */}

    // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
    // for topic performance : must send reply after process ClientMessage
    // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
    if (!not.isPersistent() && replies.size() > 0) {
      forward(from, new SendRepliesNot(replies));
    }
  }

  /**
   * Checks the reading permission of a given client agent.
   *
   * @param client  AgentId of the client requesting a reading permission.
   */
  protected boolean isReader(AgentId client) {
    if (isAdministrator(client) || freeReading)
      return true;

    Integer clientRight = (Integer) clients.get(client);
    if (clientRight == null) return false;
    
    return ((clientRight.intValue() == READ) ||
            (clientRight.intValue() == READWRITE));
  }

  /**
   * Checks the writing permission of a given client agent.
   *
   * @param client  AgentId of the client requesting a writing permission.
   */
  protected boolean isWriter(AgentId client) {
    if (isAdministrator(client) || freeWriting)
      return true;

    Integer clientRight = (Integer) clients.get(client);
    if (clientRight == null) return false;
    
    return ((clientRight.intValue() == WRITE) ||
            (clientRight.intValue() == READWRITE));
  }

  /**
   * Checks the administering permission of a given client agent.
   *
   * @param client  AgentId of the client requesting an admin permission.
   */
  protected boolean isAdministrator(AgentId client) {
    return AdminTopic.isAdminTopicId(client) || client.equals(adminId);
  }

  abstract protected void doClientMessages(AgentId from, ClientMessages not, boolean throwsExceptionOnFullDest) throws AccessException;
  abstract protected void doUnknownAgent(UnknownAgent not);
  abstract protected void doDeleteNot(DeleteNot not);
  
  public void delete() {
    DeleteDestination request = new DeleteDestination(getDestinationId());
    FwdAdminRequestNot deleteNot = new FwdAdminRequestNot(request, null, null);
    Channel.sendTo(AdminTopic.getDefault(), deleteNot);
  }

  // AF (TODO): We have to define an interface that allow subclass to declare
  // a processing through delegation.

  /**
   * This method is needed to add processing before the standard handling. It
   * is used in subclass of {@link Queue} and {@link Topic}.
   * The incoming messages can be modified or deleted during the processing.
   * 
   * @param from The sender of the message
   * @param msgs The incoming messages.
   * @return The incoming messages after processing.
   */
  protected ClientMessages preProcess(AgentId from, ClientMessages msgs) {
    // nothing to do.
    return msgs;
  }

  /**
   * This method is needed to add processing after the standard handling. It
   * is used in subclass of {@link Queue} and {@link Topic}.
   * The incoming messages can be modified or deleted during the processing.
   * 
   * @param msgs The incoming messages.
   */
  protected void postProcess(ClientMessages msgs) {
    // nothing to do.
  }
  
  private void writeObject(java.io.ObjectOutputStream out) throws IOException {
    out.writeBoolean(deletable);
    out.writeObject(adminId);
    out.writeBoolean(freeReading);
    out.writeBoolean(freeWriting);
    out.writeObject(clients);    
    out.writeObject(dmqId);
    out.writeLong(creationDate);
    out.writeLong(nbMsgsDeliverSinceCreation);
    out.writeLong(nbMsgsSentToDMQSinceCreation);
    out.writeLong(period);
    out.writeObject(interceptorsProp);
  }

  private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
    deletable = in.readBoolean();
    adminId = (AgentId) in.readObject();
    freeReading = in.readBoolean();
    freeWriting = in.readBoolean();
    clients = (Hashtable<AgentId, Integer>) in.readObject();
    dmqId = (AgentId)in.readObject();
    strbuf = new StringBuffer();
    creationDate = in.readLong();
    nbMsgsDeliverSinceCreation = in.readLong();
    nbMsgsSentToDMQSinceCreation = in.readLong();
    period = in.readLong();
    interceptorsProp = (List<Properties>) in.readObject();
  }

  // DestinationMBean interface

  /**
   * Returns the unique identifier of the destination.
   *
   * @return the unique identifier of the destination.
   */
  public final String getDestinationId() {
    return getId().toString();
  }

  /**
   * Tests if this destination is free for reading.
   *
   * @return true if anyone can receive messages from this destination;
   * 	     false otherwise.
   */
  public boolean isFreeReading() {
    return freeReading;
  }

  /**
   * Sets the <code>FreeReading</code> attribute for this destination.
   *
   * @param on	if true anyone can receive message from this destination.
   */
  public void setFreeReading(boolean on) {
    // state change, so save.
    setSave();
    freeReading = on;
  }

  /**
   * Tests if this destination is free for writing.
   *
   * @return true if anyone can send messages to this destination;
   * 	     false otherwise.
   */
  public boolean isFreeWriting() {
    return freeWriting;
  }

  /**
   * Sets the <code>FreeWriting</code> attribute for this destination.
   *
   * @param on	if true anyone can send message to this destination.
   */
  public void setFreeWriting(boolean on) {
    // state change, so save.
    setSave();
    freeWriting = on;
  }

  /**
   * Return the unique identifier of DMQ set for this destination if any.
   *
   * @return the unique identifier of DMQ set for this destination if any;
   *	     null otherwise.
   */
  public String getDMQId() {
    if (dmqId != null) 
      return dmqId.toString();
    return null;
  }

  public AgentId getDMQAgentId() {
    return dmqId;
  }

  /**
   * Returns this destination creation time as a long.
   *
   * @return the destination creation time as UTC milliseconds from the epoch.
   */
  public long getCreationTimeInMillis() {
    return creationDate;
  }

  /**
   * Returns this destination creation time through a <code>String</code> of
   * the form: <code>dow mon dd hh:mm:ss zzz yyyy</code>.
   *
   * @return the destination creation time.
   */
  public String getCreationDate() {
    return new Date(creationDate).toString();
  }

  /**
   * Returns the number of messages received since creation time of this
   * destination.
   *
   * @return the number of messages received since creation time.
   */
  abstract public long getNbMsgsReceiveSinceCreation();
  
  /**
   * Returns the number of messages delivered since creation time of this
   * destination. It includes messages all delivered messages to a consumer,
   * already acknowledged or not.
   *
   * @return the number of messages delivered since creation time.
   */
  public long getNbMsgsDeliverSinceCreation() {
    return nbMsgsDeliverSinceCreation;
  }

  /**
   * Returns the number of erroneous messages forwarded to the DMQ since
   * creation time of this destination..
   *
   * @return the number of erroneous messages forwarded to the DMQ.
   */
  public long getNbMsgsSentToDMQSinceCreation() {
    return nbMsgsSentToDMQSinceCreation;
  }

  protected void replyToTopic(AdminReply reply,
                              AgentId replyTo,
                              String requestMsgId,
                              String replyMsgId) {
    Message message = MessageHelper.createMessage(replyMsgId, requestMsgId, getAgentId(), getType());
    try {
      message.setAdminMessage(reply);
      ClientMessages clientMessages = new ClientMessages(-1, -1, message);
      forward(replyTo, clientMessages);
    } catch (Exception exc) {
      if (logger.isLoggable(BasicLevel.ERROR))
        logger.log(BasicLevel.ERROR, "", exc);
      throw new Error(exc.getMessage());
    }
  }
  
  protected final void forward(AgentId to, Notification not) {
    Channel.sendTo(to, not);
  }
  
  /**
   * 
   * @param from
   * @param not
   */
  protected void handleAdminRequestNot(AgentId from, FwdAdminRequestNot not) {
    AdminRequest adminRequest = not.getRequest();

    if (adminRequest instanceof SetRight) {
      setRight((SetRight) adminRequest,
               not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof GetStatsRequest) {
      String names = ((GetStatsRequest) adminRequest).getAttributes();
      replyToTopic(new GetStatsReply(getJMXStatistics(names)),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof SetDMQRequest) {
      // state change, so save.
      setSave();

      if (((SetDMQRequest)adminRequest).getDmqId() != null)
        dmqId = AgentId.fromString(((SetDMQRequest)adminRequest).getDmqId());
      else
        dmqId = null;

      replyToTopic(new AdminReply(true, null),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof AdminCommandRequest) {
      processAdminCommand((AdminCommandRequest) adminRequest, not.getReplyTo(), not.getRequestMsgId());
    } else {
      logger.log(BasicLevel.ERROR, "Unknown administration request for destination " + getId());
      replyToTopic(new AdminReply(AdminReply.UNKNOWN_REQUEST, null),
                   not.getReplyTo(),
                   not.getRequestMsgId(),
                   not.getReplyMsgId());
      
    }
  }
  
  /**
   * Proccess an admin command.
   * 
   * @param request The administration request.
   * @param replyTo The destination to reply.
   * @param requestMsgId The JMS message id needed to reply.
   */
  protected void processAdminCommand(AdminCommandRequest request,	AgentId replyTo, String requestMsgId) {
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "processAdminCommand(" + request + ", " + replyTo + ", " + requestMsgId + ')');
  	Properties prop = null;
  	Properties replyProp = null;
  	try {
			switch (request.getCommand()) {
			case AdminCommandConstant.CMD_ADD_INTERCEPTORS:
				prop = request.getProp();
				addInterceptor(getDestinationId(), getName(), prop);
				break;
				
			case AdminCommandConstant.CMD_REMOVE_INTERCEPTORS:
				prop = request.getProp();
				String classNames = (String) prop.get(AdminCommandConstant.INTERCEPTORS);
				removeInterceptor(classNames);
				break;
				
			case AdminCommandConstant.CMD_GET_INTERCEPTORS:
			  replyProp = new Properties();
			  if (interceptors == null) {
			    replyProp.put(AdminCommandConstant.INTERCEPTORS, "");
			  } else {
			    replyProp.put(AdminCommandConstant.INTERCEPTORS, InterceptorsHelper.getListInterceptors(interceptors));
			  }
			  break;
			  
			case AdminCommandConstant.CMD_REPLACE_INTERCEPTORS:
				prop = request.getProp();
				if (interceptors == null)
					throw new Exception("interceptors == null.");
				replaceInterceptor(getDestinationId(), getName(), prop);
				break;
				
			case AdminCommandConstant.CMD_SET_PROPERTIES:
				setProperties(request.getProp(), false);
				setSave();
				break;
			case AdminCommandConstant.CMD_START_HANDLER:
				replyProp = processStartHandler(request.getProp());
				break;
			case AdminCommandConstant.CMD_STOP_HANDLER:
				replyProp = processStopHandler(request.getProp());
				break;
			case AdminCommandConstant.CMD_PAUSE:
			  replyProp = processPause(request.getProp());
			  break;
			default:
				throw new Exception("Bad command : \"" + request.getCommand() + "\"");
			}
			// reply
			replyToTopic(new AdminCommandReply(true, AdminCommandConstant.getCommandNames(request.getCommand()) + " done.", replyProp), replyTo, requestMsgId, requestMsgId);
		} catch (Exception exc) {
			if (logger.isLoggable(BasicLevel.WARN))
				logger.log(BasicLevel.WARN, "", exc);
			replyToTopic(new AdminReply(-1, exc.getMessage()), replyTo, requestMsgId, requestMsgId);
		}
  }
  
  /**
   * Start the acquisition queue/topic handler.
   * 
   * @param prop properties for start if needed (can be null)
   * @return properties for the reply.
   * @throws Exception
   */
  protected Properties processStartHandler(Properties prop) throws Exception { 
  	if (this instanceof AcquisitionQueue) {
  		return ((AcquisitionQueue) this).startHandler(prop);
  	} else if (this instanceof AcquisitionTopic) {
  		return ((AcquisitionTopic) this).startHandler(prop);
  	} else {
  		throw new Exception("processStartHandler :: bad destination.");
  	}
  }
  
  /**
   * Stop the acquisition queue/topic handler.
   * 
   * @param prop properties for start if needed (can be null)
   * @return properties for the reply.
   * @throws Exception
   */
  protected Properties processStopHandler(Properties prop) throws Exception { 
  	if (this instanceof AcquisitionQueue) {
  		return ((AcquisitionQueue) this).stopHandler(prop);
  	} else if (this instanceof AcquisitionTopic) {
  		return ((AcquisitionTopic) this).stopHandler(prop);
  	} else {
  		throw new Exception("processStopHandler :: bad destination.");
  	}
  }
  
  /**
   * Pause / Resume the message distribution.
   * 
   * @param prop properties for start if needed (can be null)
   * @return properties for the reply.
   * @throws Exception
   */
  protected Properties processPause(Properties prop) throws Exception {
    boolean pause = Boolean.parseBoolean(prop.getProperty(AdminCommandConstant.PAUSE));
    if (this instanceof Queue) {
      ((Queue) this).setPause(pause);
      return prop;
    } else {
      throw new Exception("processPause :: not a Queue.");
    }
  }

  /**
   * Run all defined interceptors on the message.
   * 
   * @param msg the message
   * @return message potentially modified by the interceptors, null if an interceptor reject the message.
   */
  protected Message processInterceptors(Message msg) {
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "processInterceptors(" + msg + ')');
  	
  	if (interceptors != null && !interceptors.isEmpty()) {
  		Iterator<MessageInterceptor> it = interceptors.iterator();
  		while (it.hasNext()) {
  			MessageInterceptor interceptor = (MessageInterceptor) it.next();
  			if (!interceptor.handle(msg, -1))
  				return null;
  		}
  	}
  	return msg;
  }
  
  /**
   * @return true if interceptors set
   */
  protected boolean interceptorsAvailable() {
  	return interceptors != null && !interceptors.isEmpty();
  }
  
  private void addInterceptor(String agentId, String agentName, final Properties prop) throws Exception {
    String error = null;
    String interceptorsClassName = prop.getProperty(AdminCommandConstant.INTERCEPTORS);
    if (interceptorsClassName == null) return;
    
    if (interceptorsProp == null)
      interceptorsProp = new ArrayList<Properties>();
    if (interceptors == null)
      interceptors = new ArrayList<MessageInterceptor>();
    
    if (interceptorsClassName.contains(InterceptorsHelper.INTERCEPTOR_CLASS_NAME_SEPARATOR)) {
      StringTokenizer token = new StringTokenizer(interceptorsClassName, InterceptorsHelper.INTERCEPTOR_CLASS_NAME_SEPARATOR);
      while (token.hasMoreTokens()) {
        String interceptorClassName = token.nextToken();
        Properties iProp = new Properties();
        iProp.setProperty(AdminCommandConstant.INTERCEPTORS, interceptorClassName);
        interceptorsProp.add(iProp);
        try {
          InterceptorsHelper.addInterceptors(agentId, agentName, AdminCommandConstant.INTERCEPTORS, iProp, interceptors);
        } catch(Exception exc) {
          if (logger.isLoggable(BasicLevel.WARN))
            logger.log(BasicLevel.WARN, "addInterceptors", exc);
          StringWriter sw = new StringWriter();
          exc.printStackTrace(new PrintWriter(sw));
          if (error == null)
          error = "(" + interceptorClassName + " exc=" + sw.toString() + ')';
          else 
            error += "(" + interceptorClassName + " exc=" + sw.toString() + ')';
        }
      }
    } else {
      interceptorsProp.add(prop);
      InterceptorsHelper.addInterceptors(agentId, agentName, AdminCommandConstant.INTERCEPTORS, prop, interceptors);
    }
    
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "addInterceptor interceptors = " + interceptors + ", interceptorsProp = " + interceptorsProp);

    // state change
    setSave();
    
    if (error != null) {
      if (logger.isLoggable(BasicLevel.WARN))
        logger.log(BasicLevel.WARN, "addInterceptor error = " + error);
      throw new Exception(error);
    }
  }
  
  private void removeInterceptor(String classNames) throws Exception {
    if (interceptorsProp == null) 
      return;
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "removeInterceptor classeNames = " + classNames + ", interceptors = " + interceptors + ", interceptorsProp = " + interceptorsProp);

    StringTokenizer token = new StringTokenizer(classNames, InterceptorsHelper.INTERCEPTOR_CLASS_NAME_SEPARATOR);
    while (token.hasMoreTokens()) {
      String className = token.nextToken();
      // find and remove the interceptor
      Properties toRemove = null;
      Iterator<Properties> it = interceptorsProp.iterator();
      while (it.hasNext()) {
        Properties properties = (Properties) it.next();
        if (properties.getProperty(AdminCommandConstant.INTERCEPTORS).equals(className)) {
          toRemove = properties;
          break;
        }
      }
      if (toRemove != null) {
        interceptorsProp.remove(toRemove);
        
        InterceptorsHelper.removeInterceptors(classNames, interceptors);
        if (interceptors.isEmpty())
          interceptors = null;
        // state change
        setSave();
      }
    }
  }
  
  private void replaceInterceptor(String agentId, String agentName, final Properties prop) throws Exception {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "replaceInterceptor(" + agentId + ", " + agentName + ", " + prop + ')');
    
    Boolean ret = InterceptorsHelper.replaceInterceptor(
        agentId, 
        agentName, 
        AdminCommandConstant.INTERCEPTORS_NEW, 
        AdminCommandConstant.INTERCEPTORS_OLD, 
        interceptors, 
        prop);
    
    if (ret) {
      // update the interceptorsProp list
      Iterator<Properties> it = interceptorsProp.iterator();
      while (it.hasNext()) {
        Properties properties = (Properties) it.next();
        if (properties.getProperty(AdminCommandConstant.INTERCEPTORS).equals(prop.getProperty(AdminCommandConstant.INTERCEPTORS_OLD))) {
          int index = interceptorsProp.indexOf(properties);
          Properties newProp = new Properties();
          Set<Entry<Object, Object>> entrys = prop.entrySet();
          Iterator<Entry<Object, Object>> iterator = entrys.iterator();
          while (iterator.hasNext()) {
            Entry<Object, Object> entry = (Entry<Object, Object>) iterator.next();
            if (entry.getKey().equals(AdminCommandConstant.INTERCEPTORS_NEW))
              newProp.put(AdminCommandConstant.INTERCEPTORS, entry.getValue());
            else
              newProp.put(entry.getKey(), entry.getValue());
          }
          interceptorsProp.remove(index);
          interceptorsProp.add(index, newProp);
          // state change
          setSave();
          break;
        }
      }
    }
  }
  
  // Encodable interface

  public int getEncodedSize() throws Exception {
    int encodedSize = super.getEncodedSize();
    encodedSize += BOOLEAN_ENCODED_SIZE;
    encodedSize += adminId.getEncodedSize();
    encodedSize += BOOLEAN_ENCODED_SIZE + BOOLEAN_ENCODED_SIZE + INT_ENCODED_SIZE;
    Iterator<Entry<AgentId, Integer>> clientIterator = clients.entrySet().iterator();
    while (clientIterator.hasNext()) {
      Entry<AgentId, Integer> client = clientIterator.next();
      encodedSize += client.getKey().getEncodedSize();
      encodedSize += INT_ENCODED_SIZE;
    }
    encodedSize += BOOLEAN_ENCODED_SIZE;
    if (dmqId != null) {
      encodedSize += dmqId.getEncodedSize();
    }
    encodedSize += LONG_ENCODED_SIZE * 4;
    
    encodedSize += BOOLEAN_ENCODED_SIZE;
    if (interceptorsProp != null) {
      encodedSize += INT_ENCODED_SIZE;
      for (Properties properties : interceptorsProp) {
        encodedSize += EncodableHelper.getEncodedSize(properties);
      }
    }

    return encodedSize;
  }
  
  public void encode(Encoder encoder) throws Exception {
    super.encode(encoder);
    encoder.encodeBoolean(deletable);
    adminId.encode(encoder);
    encoder.encodeBoolean(freeReading);
    encoder.encodeBoolean(freeWriting);
    encoder.encodeUnsignedInt(clients.size());
    Iterator<Entry<AgentId, Integer>> clientIterator = clients.entrySet().iterator();
    while (clientIterator.hasNext()) {
      Entry<AgentId, Integer> client = clientIterator.next();
      client.getKey().encode(encoder);
      encoder.encodeUnsignedInt(client.getValue());
    }
    if (dmqId == null) {
      encoder.encodeBoolean(true);
    } else {
      encoder.encodeBoolean(false);
      dmqId.encode(encoder);
    }
    encoder.encodeUnsignedLong(creationDate);
    encoder.encodeUnsignedLong(nbMsgsDeliverSinceCreation);
    encoder.encodeUnsignedLong(nbMsgsSentToDMQSinceCreation);
    encoder.encodeUnsignedLong(period);

    if (interceptorsProp == null) {
      encoder.encodeBoolean(true);
    } else {
      encoder.encodeBoolean(false);
      encoder.encodeUnsignedInt(interceptorsProp.size());
      for (Properties properties : interceptorsProp) {
        EncodableHelper.encodeProperties(properties, encoder);
      }
    }
  }

  public void decode(Decoder decoder) throws Exception {
    super.decode(decoder);
    deletable = decoder.decodeBoolean();
    adminId = new AgentId((short) 0, (short) 0, 0);
    adminId.decode(decoder);
    freeReading = decoder.decodeBoolean();
    freeWriting = decoder.decodeBoolean();
    int clientsSize = decoder.decodeUnsignedInt();
    clients = new Hashtable<AgentId, Integer>(clientsSize);
    for (int i = 0; i < clientsSize; i++) {
      AgentId key = new AgentId((short) 0, (short) 0, 0);
      key.decode(decoder);
      int value = decoder.decodeUnsignedInt();
      clients.put(key, value);
    }
    boolean isNull = decoder.decodeBoolean();
    if (isNull) {
      dmqId = null;
    } else {
      dmqId = new AgentId((short) 0, (short) 0, 0);
      dmqId.decode(decoder);
    }
    creationDate = decoder.decodeUnsignedLong();
    nbMsgsDeliverSinceCreation = decoder.decodeUnsignedLong();
    nbMsgsSentToDMQSinceCreation = decoder.decodeUnsignedLong();
    period = decoder.decodeUnsignedLong();
    
    isNull = decoder.decodeBoolean();
    if (isNull) {
      interceptorsProp = null;
    } else {
      int size = decoder.decodeUnsignedInt();
      interceptorsProp = new ArrayList<Properties>();
      for (int i = 0; i < size; i++) {
        Properties properties = EncodableHelper.decodeProperties(decoder);
        interceptorsProp.add(properties);
      }
    }
  }
  
}
