/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2006 - 2025 ScalAgent Distributed Technologies
 *
 * 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): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package org.objectweb.joram.mom.messages;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.ref.SoftReference;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Vector;

import org.objectweb.joram.mom.util.JoramHelper;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.AgentServer;
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;
import fr.dyade.aaa.common.stream.StreamUtil;
import fr.dyade.aaa.util.Transaction;

/** 
 * The <code>Message</code> class actually provides the transport facility
 * for the data exchanged during MOM operations.
 * <p>
 * A message content is always wrapped as a bytes array, it is characterized
 * by properties and "header" fields.
 */
public final class Message implements Comparable<Message>, Serializable, Encodable, MessageView {
  /** define serialVersionUID for interoperability */
  private static final long serialVersionUID = 2L;

  /** logger */
  private static final Logger logger = Debug.getLogger(Message.class.getName());

  /** Arrival position of this message on its queue or proxy. */
  transient public long order;

  @Override
  public int compareTo(Message msg) {
    // This method allows to sort messages list in the history order (see JORAM-358).
    // Be careful, specific comparators are using in loadAll method.
    return Long.compare(order, msg.order);
  }

  /**
   * The number of acknowledgements a message still expects from its 
   * subscribers before having been fully consumed by them (field used
   * by JMS proxies).
   * Be careful, this field is not saved but set to 0 during message
   * loading then calculated during the proxy initialization.
   */
  public transient int acksCounter;

  /**
   * The number of acknowledgements a message still expects from its 
   * durable subscribers before having been fully consumed by them (field used
   * by JMS proxies).
   * Be careful, this field is not saved but set to 0 during message
   * loading then calculated during the proxy initialization.
   */
  public transient int durableAcksCounter;
  
  /**
   * Reference to the MOM message.
   * Should not be used directly, rather use getMsg so the message is restored from
   * disk if necessary (currently not used)
   */
  private transient org.objectweb.joram.shared.messages.Message msg;
  
  /**
   * Gets the message, loads it from disk if needed, then sets the msg attribute.
   * 
   * TODO (AF): /!\ Be careful, previously this method can return null if the message cannot be 
   * loaded from disk. Returning a specific RuntimeException is a work-around to avoid NPE. In a
   * best world this exception should be handled at higher level.
   * 
   * @return the message.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public final org.objectweb.joram.shared.messages.Message getSharedMsg() throws InvalidMessageException {
    if (msg == null) {
      // If the message is not loaded, load it from disk (JORAM-358).
      // Currently should never happened.
      try {
        Message m = load(txname);
        msg = m.msg;
      } catch (ClassNotFoundException | IOException exc) {
        logger.log(BasicLevel.ERROR, "Cannot load message " + txname, exc);
        throw new InvalidMessageException("Cannot get message from disk: " + txname);
      }
    }
    return msg;
  }

  /**
   *  Defines if the queue restoration at startup must use the loadAll transaction
   * mechanism when it is implemented. 
   * <p>
   *  Default value is false.
   * <p>
   *  This property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  private static final boolean useLoadALL = AgentServer.getBoolean(org.objectweb.joram.shared.messages.Message.BROKER_USELOADALL);
  
  /** SoftReference to the body of the MOM message. */
  private transient SoftReference<byte[]> bodySoftRef = null;

  /**
   * <code>true</code> if soft reference can be used for the message.
   */
  private transient boolean soft;

  /**
   *  Defines if the swapping mechanism is globally activated for messages
   * in this server.
   * <p>
   *  Default value is false.
   * <p>
   *  Note: the message swapping can be finely configured using the 
   * <code>JMS_JORAM_SWAPALLOWED</code> property of the JMS message.
   * <p>
   *  This property can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  private static final boolean globalUseSoftRef = AgentServer.getBoolean(org.objectweb.joram.shared.messages.Message.BROKER_SWAPALLOWED);
  
  /**
   * Empty constructor.
   */
  public Message() {}

  /**
   * Constructs a <code>Message</code> instance.
   */
  public Message(org.objectweb.joram.shared.messages.Message msg) {
    this.msg = msg;
    // Soft reference can be used only if message is persistent and has a body.
    Boolean msgUseSoftRef = (Boolean) msg.getProperty(org.objectweb.joram.shared.messages.Message.SWAPALLOWED);
    if (msgUseSoftRef != null) {
      this.soft = msgUseSoftRef.booleanValue() && msg.persistent && !msg.isNullBody();
    } else {
      this.soft = globalUseSoftRef && msg.persistent && !msg.isNullBody();
    }
  }

  /**
   * Returns the contained message eventually without the body.
   * 
   * @return The contained message.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public org.objectweb.joram.shared.messages.Message getHeaderMessage() throws InvalidMessageException {
    // Loads the message if needed.
    getSharedMsg();
    
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "MessagePersistenceModule.getHeaderMessage() -> " + msg);
    
    return msg;
  }

  /**
   * Gets the body of message, and locks it in memory.
   * If needed the body is loaded from persistence.
   */
  private void getBody() {
    // The message is not soft, or body is already lock in memory!
    if (!soft || !msg.isNullBody())
      return;

    // Try to get the body from soft reference.
    if (bodySoftRef != null) {
      msg.setRawBody(bodySoftRef.get());
      bodySoftRef = null;
      // If the body is always in SoftRef returns
      if (!msg.isNullBody()) return;
    }

    // If the body is no longer in SoftRef, try to load the body from repository
    byte[] bytes = loadBody(AgentServer.getTransaction(), txname);
    msg.setBody(bytes, 0, -1);
  }
  
  private static byte[] loadBody(Transaction transaction, String txname) {
    byte[] bytes = null;
    try {
      bytes = transaction.loadByteArray(txname + "B");
    } catch (Exception exc) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR, "Body of message named [" + txname + "] could not be loaded", exc);
      else
        logger.log(BasicLevel.ERROR, "Body of message named [" + txname + "] could not be loaded: " + exc);
    }
    return bytes;
  }
  
  /**
   * Releases the body of message.
   * 
   * This body is only accessible via a soft reference, garbage can collect this object if necessary.
   * If soft is false, or body is null, nothing happens.
   */
  private void releaseBody() {
    if (soft && (!msg.isNullBody())) {
      bodySoftRef = new SoftReference<byte[]>(msg.getRawBody());
      msg.setRawBody(null);
    }
  }
  
  /**
   * Returns the contained message with body.
   * 
   * If needed the body is loaded from repository.
   * 
   * @return The contained message.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public org.objectweb.joram.shared.messages.Message getFullMessage() throws InvalidMessageException {
    getSharedMsg();
    
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "MessagePersistenceModule.getFullMessage() " + txname);
    // The message can be soft but releaseFullMessage has not been called !

    getBody();
    return msg;
  }

  /**
   * Creates a soft reference instead of a hard one linking to the body of the
   * contained message. The message must have been saved previously.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void releaseFullMessage() throws InvalidMessageException {
    getSharedMsg();
    releaseBody();
  }

  /**
   * Returns the message type. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public int getType() throws InvalidMessageException {
    return getSharedMsg().type;
  }

  /**
   * Returns the message identifier. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public String getId() throws InvalidMessageException {
    return getSharedMsg().id;
  }

  /**
   * Sets the message identifier. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */ 
  public void setIdentifier(String id) throws InvalidMessageException {
    getSharedMsg().id = id;
  }

  /**
   * Returns <code>true</code> if the message is persistent. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public boolean isPersistent() throws InvalidMessageException {
    return getSharedMsg().persistent;
  }

  /**
   * Sets the message persistence mode. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setPersistent(boolean persistent) throws InvalidMessageException {
    getSharedMsg().persistent = persistent;
  }

  /**
   * Returns the message priority. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public int getPriority() throws InvalidMessageException {
    return getSharedMsg().priority;
  }

  /**
   * Sets the message priority.
   *
   * @param priority  Priority value: 0 the lowest, 9 the highest, 4 normal.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */ 
  public void setPriority(int priority) throws InvalidMessageException {
    if (priority >= 0 && priority <= 9)
      getSharedMsg().priority = priority;
  }

  /**
   * Returns the message expiration time. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public long getExpiration() throws InvalidMessageException {
    return getSharedMsg().expiration;
  }

  /**
   * Sets the message expiration.
   *
   * @param expiration	The expiration time.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setExpiration(long expiration) throws InvalidMessageException {
    if (expiration >= 0)
      getSharedMsg().expiration = expiration;
  }

  /**
   * Returns the message time stamp. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public long getTimestamp() throws InvalidMessageException {
    return getSharedMsg().timestamp;
  }

  /**
   * Sets the message time stamp. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setTimestamp(long timestamp) throws InvalidMessageException {
    getSharedMsg().timestamp = timestamp;
  }

  /**
   * Returns the message correlation identifier. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public final String getCorrelationId() throws InvalidMessageException {
    return getSharedMsg().correlationId;
  }

  /**
   * Sets the message correlation identifier. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setCorrelationId(String correlationId) throws InvalidMessageException {
    getSharedMsg().correlationId = correlationId;
  }

  /**
   * Returns the message delivery count.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public int getDeliveryCount() throws InvalidMessageException {
    return getSharedMsg().deliveryCount;
  }

  /**
   * Sets the message delivery count. 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setDeliveryCount(int deliveryCount) throws InvalidMessageException {
    getSharedMsg().deliveryCount = deliveryCount;
  }

  /**
   * Increments the message delivery count.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void incDeliveryCount() throws InvalidMessageException {
    getSharedMsg().deliveryCount += 1;
  }

  /**
   * Decrements the message delivery count.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void decDeliveryCount() throws InvalidMessageException {
    getSharedMsg().deliveryCount -= 1;
  }

  /**
   * Sets the message redelivered flag.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setRedelivered() throws InvalidMessageException {
    getSharedMsg().redelivered = true;
  }
  
  /**
   * Sets the message delivery time 
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setDeliveryTime(long deliveryTime) throws InvalidMessageException {
    getSharedMsg().deliveryTime = deliveryTime;
  }
  
  /**
   * Returns the message delivery time
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public long getDeliveryTime() throws InvalidMessageException {
    return getSharedMsg().deliveryTime;
  }
  
  public long getOrder() {
    return order;
  }

  public void setOrder(long order) {
    this.order = order;
  }

  public int getAcksCounter() {
    return acksCounter;
  }

  public void incAcksCounter() {
    this.acksCounter++;
  }

  public int getDurableAcksCounter() {
    return durableAcksCounter;
  }

  public void incDurableAcksCounter() {
    this.durableAcksCounter++;
  }

  /**
   * Get the clientID
   * @return the clientID
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public String getClientID() throws InvalidMessageException {
    return getSharedMsg().clientID;
  }

  /**
   * Sets a property value.
   * If the value is not a Java primitive object its string representation is used.
   *
   * @param name  The property name.
   * @param value  The property value.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public void setObjectProperty(String name, Object value) throws InvalidMessageException {
    getSharedMsg().setProperty(name, value);
  }

  //   /**
  //    * Sets an object as the body of the message. 
  //    * AF: Used to wrap administration message !!
  //    *
  //    * @exception IOException  In case of an error while setting the object.
  //    */
  //   public void setObject(Object object) throws IOException {
  //     msg.type = Message.OBJECT;

  //     if (object == null) {
  //       msg.body = null;
  //     } else {
  //       ByteArrayOutputStream baos = new ByteArrayOutputStream();
  //       ObjectOutputStream oos = new ObjectOutputStream(baos);
  //       oos.writeObject(object);
  //       oos.flush();
  //       msg.body = baos.toByteArray();
  //       oos.close();
  //       baos.close();
  //     }
  //   }

  //   /**
  //    * Returns the object body of the message.
  //    * AF: Used to wrap administration message !!
  //    *
  //    * @exception IOException  In case of an error while getting the object.
  //    * @exception ClassNotFoundException  If the object class is unknown.
  //    */
  //   public Object getObject() throws Exception {
  //     // AF: May be, we should verify that it is an Object message!!
  //     if (msg.body == null) return null;

  //     ByteArrayInputStream bais = new ByteArrayInputStream(msg.body);
  //     ObjectInputStream ois = new ObjectInputStream(bais);
  //     Object obj = ois.readObject();
  //     ois.close();

  //     return obj;
  //   }

  /**
   * Returns <code>true</code> if the message is valid.
   * The message is valid if not expired.
   *
   * @param currentTime	The current time to verify the expiration time.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public boolean isValid(long currentTime) throws InvalidMessageException {
    return (getSharedMsg().expiration <= 0) || (getSharedMsg().expiration > currentTime);
  }

  /**
   * Return true if the message has an expiration delay.
   * 
   * @return true if the message has an expiration delay.
   * @throws InvalidMessageException The message cannot be loaded from disk.
   */
  public final boolean hasExpiration() throws InvalidMessageException {
    return getSharedMsg().expiration > 0;
  }
  
  /** Name used to store the message */
  transient String txname = null;

  public final void setTxName(String txname) {
    this.txname = txname;
  }

  public final String getTxName() {
    return txname;
  }

  public static Message load(String txname) throws IOException, ClassNotFoundException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Message.load:" + txname);
    Message m = (Message) AgentServer.getTransaction().load(txname);
    m.txname = txname;
    return m;
  }

  /**
   * Method used to save the initial state of the message.
   * Should be called only once for each message.
   */
  public void save() {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Message.save:" + txname);

    try {
      // Calls getSharedMessage, and then loads the message if needed
      if (!isPersistent()) return;

      if (soft) {
        byte[] body = msg.getRawBody();
        try {
          // sets the body to null to save it in an other file
          msg.setRawBody(null);
          AgentServer.getTransaction().create(this, txname);
        } finally {
          msg.setRawBody(body);
        }
        // Save the body, The body is RO do not copy it (Create operation).
        AgentServer.getTransaction().saveByteArray(body, null, txname + "B", false, true);
      } else {
        // Saves the message with its body.
        AgentServer.getTransaction().create(this, txname);
      }
    } catch (IOException exc) {
      // TODO (AF): Should never happened, may be we should rethrow the exception
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR, "Message named [" + txname + "] cannot be saved", exc);
      else
        logger.log(BasicLevel.ERROR, "Message named [" + txname + "] cannot be saved: " + exc.getMessage());
    }
  }

  /**
   * Method used to save the header of a message after modification.
   * The body of a message should never be saved.
   */
  public void saveHeader() {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Message.saveHeader:" + txname);

    try {
      // Calls getSharedMessage, and then loads the message if needed
      if (!isPersistent()) return;

      if (soft) {
        byte[] body = msg.getRawBody();
        try {
          // sets the body to null to save it in an other file
          msg.setRawBody(null);
          AgentServer.getTransaction().save(this, txname);
        } finally {
          msg.setRawBody(body);
        }
      } else {
        // Saves the message with its body.
        AgentServer.getTransaction().save(this, txname);
      }
    } catch (IOException exc) {
      // TODO (AF): Should never happened, may be we should rethrow the exception
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR, "Message named [" + txname + "] cannot be saved", exc);
      else
        logger.log(BasicLevel.ERROR, "Message named [" + txname + "] cannot be saved: " + exc.getMessage());
    }
  }

  public void delete() {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Message.delete:" + txname);

    // Calls getSharedMessage, and then loads the message if needed
    if (!isPersistent()) return;

    AgentServer.getTransaction().delete(txname);
    if (soft)
      AgentServer.getTransaction().delete(txname + "B");
  }

  /**
   * Loads all persisted messages.
   * The returned list is sorted according to the message order and priority if any.
   * 
   * If there is more message to restore than max, the Message structure is created but the
   * message is not loaded. Currently this method is always called with Integer.MAX_VALUE so
   * all messages are really loaded.
   * 
   * @param msgTxname prefix of messages to load.
   * @param max the maximum number of messages to load (currently unused).
   * @return a vector containing the loaded messages.
   */
  public static Vector<Message> loadAll(String msgTxname, int max) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Message.loadAll() " + msgTxname);

    Vector<Message> messages = new Vector<Message>();
    Transaction tx = AgentServer.getTransaction();

    // TODO (AF): Use DBTransaction optimized loadAll method to get all messages from a unique request.
    // Currently this optimization results in an issue if messages swap is allowed, it needs to filter
    // body objects that have the same prefix).
    if (tx.useLoadAll() && useLoadALL) {
      // Retrieving the saved messages from persistency. 
      long start = System.currentTimeMillis();
      Map<String, Message> msgsMap = new HashMap<String, Message>(32000);
      tx.loadAll(msgTxname + "", msgsMap);
      long end = System.currentTimeMillis();

      logger.log(BasicLevel.INFO,
                 "Message.loadAll: all messages loaded (" + msgsMap.size() + ") -> " + (end - start));

      start = System.currentTimeMillis();
      for (Map.Entry<String, Message> entry : msgsMap.entrySet()) {
        String name = entry.getKey();
        if (name.charAt(name.length() - 1) == 'B')
          // Never happen, the corresponding objects are bytes arrays and throw exceptions in loadAll
          continue;
        Message msg = entry.getValue();
        msg.txname = name;
        if (msg.msg == null) {
          // TODO (AF): Currently, could not happen. The 'shared message' is stored in the 'mom message'
        }
        messages.add(msg);
      }
      end = System.currentTimeMillis();
      logger.log(BasicLevel.INFO,
                 "Message.loadAll: insert (" + messages.size() + ") -> " + (end - start));

      start = System.currentTimeMillis();
      // Sort messages list in reverse order to optimize queue restoration (JORAM-358)
      Collections.sort(messages);
      end = System.currentTimeMillis();
      logger.log(BasicLevel.INFO,
                 "Message.loadAll: sort -> " + (end - start));
    } else {
      // Retrieving the names of the persistence message previously saved. 
      long start = System.currentTimeMillis();
      String[] names = tx.getList(msgTxname);
      long end = System.currentTimeMillis();
      logger.log(BasicLevel.INFO,
                 "Message.loadAll: retrieves messages list (" + names.length + ") -> " + (end - start));
      logger.log(BasicLevel.DEBUG,
                 "Message.loadAll: names = " + Arrays.toString(names));

      // Retrieving the messages individually persisted.
      int nb = 0;
      Message msg;
      int priority = -1;
      boolean samePriorities = true;
      start = System.currentTimeMillis();
      for (String name : names) {
        if (name.charAt(name.length() - 1) == 'B')
          continue;

        try {
          // Loads the message if the number of messages loaded is less than the maximum allowed.
          // Note (AF): Currently unused. Is it useful? Should be removed.
          if (nb < max) {
            msg = (Message) tx.load(name);
            if (msg.msg.priority != priority) {
              if (priority == -1) {
                // 1st message
                priority = msg.msg.priority;
              } else {
                samePriorities = false;
              }
            }
            if (logger.isLoggable(BasicLevel.DEBUG)) {
              // Verify that the order coded in txname is the right one.
              int idx = name.lastIndexOf('_');
              long order = Long.parseLong(name.substring(idx+1));
              if (order != msg.order)
                logger.log(BasicLevel.WARN,
                           "Message.loadAll: Message " + name + " -> " + msg.order + " != " + order);
            }
          } else {
            // Note (AF): Currently unused.
            msg = new Message();
            // Get order from txname
            int idx = name.lastIndexOf('_');
            msg.order = Long.parseLong(name.substring(idx+1));
          }
          msg.txname = name;
          nb += 1;

          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG,
                       "Message.loadAll: names[" + nb + "] = " + msg.txname);
          messages.add(msg);
        } catch (Exception exc) {
          logger.log(BasicLevel.ERROR,
                     "Message.loadAll: Message named [" + name + "] can not be loaded", exc);
        }
      }
      end = System.currentTimeMillis();
      logger.log(BasicLevel.INFO,
                 "Message.loadAll: all messages loaded -> " + (end - start));

      start = System.currentTimeMillis();
      // Sort messages list in reverse order to optimize queue restoration (JORAM-358)
      if (samePriorities) {
        Collections.sort(messages, new Comparator<Message>() {
          @Override
          public int compare(Message msg1, Message msg2) {
            // TODO (AF): should we take care to null value for msg1 and/or msg2?
            return Long.compare(msg1.order, msg2.order);
          }
        });
      } else {
        Collections.sort(messages, new Comparator<Message>() {
          // TODO (AF): DO we should take in account the message priority?
          @Override
          public int compare(Message msg1, Message msg2) {
            if ((msg1.msg != null) && (msg2.msg != null)) {
              // Note (AF): Always true!!
              return Long.compare(msg1.order, msg2.order);
            }
            if ((msg1.msg != null) && (msg2.msg == null)) {
              return -1;
            }
            if ((msg1.msg == null) && (msg2.msg != null)) {
              return 1;
            }
            // TODO (AF): Strange, msg1 and msg2 should be null!
            return Long.compare(msg1.order, msg2.order);
          }
        });
      }
      end = System.currentTimeMillis();
      logger.log(BasicLevel.INFO,
                 "Message.loadAll: sort -> " + (end - start));
    }
    
    return messages;
  }

  /** Deletes all persisted objects. */
  public static void deleteAll(String msgTxname) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "MessagePersistenceModule.deleteAll() " + msgTxname);

    Transaction tx = AgentServer.getTransaction();

    // Retrieving the names of the persistence message previously saved. 
    String[] names = tx.getList(msgTxname);

    // Deleting the message.
    for (int i = 0; i < names.length; i++) {
      tx.delete(names[i]);
      tx.delete(names[i] + "B");
    }
  }
  
  // ================================================================================
  // Specific method needed by MessageView interface.

  /**
   * 
   * TODO: Only used by MOMCommandsImpl, to remove.
   */
  @Deprecated
  public String bodyToJSon() {
    try {
      getSharedMsg();
      getBody();

      StringBuilder strbuf = new StringBuilder();
      byte[] body = msg.getBody();
      // getBody trims the byte array and uncompresses it if needed
      msg.bodyToJSon(strbuf, msg.id, msg.type, body, 0, body.length, false, false);
      return strbuf.toString();
    } catch (Exception e) {
    	if (logger.isLoggable(BasicLevel.INFO))
        logger.log(BasicLevel.INFO, "getText()", e);
	    return null;
    }
  }

  public boolean isRedelivered() throws InvalidMessageException {
    return getSharedMsg().redelivered;
  }

  public Map<String, String> getProperties() throws InvalidMessageException {
    if (getSharedMsg().properties == null)
      return null;

    Map<String, String> props = new HashMap<String, String>();
    Enumeration enu = msg.properties.keys();
    while (enu.hasMoreElements()) {
      String key = (String) enu.nextElement();
      props.put(key, msg.properties.get(key).toString());
    }
    return props;
  }
  
  // ================================================================================
  // toString.
  
  public final String toString() {
    StringBuffer strbuf = new StringBuffer();
    toString(strbuf);
    return strbuf.toString();
  }

  public void toString(StringBuffer strbuf) {
    strbuf.append('(').append(super.toString());
    strbuf.append(",order=").append(order);
    strbuf.append(",soft=").append(soft);
    strbuf.append(",txname=").append(txname);
    strbuf.append(",msg=").append(msg);
    strbuf.append(')');
  }

  // ================================================================================
  // Handling export of message in JSon format.
  
//  public String toJSon(boolean binary) throws IOException {
//    StringBuilder strbuf = new StringBuilder();
//    toJSon(strbuf, binary);
//    return strbuf.toString();
//  }
  
  public void toJSon(Appendable appendable, boolean binary) throws IOException {
    appendable.append("{\n\"order\": ").append(Long.toString(order)).append(",\n");
    getFullMessage();
    msg.headerToJSon(appendable);
    appendable.append(",\n");
    byte[] body = msg.getBody();
    // getBody trims the byte array and uncompresses it if needed
    msg.bodyToJSon(appendable, msg.id, msg.type, body, 0, body.length, false, binary);
    appendable.append("}\n");
  }
  
  public void exportJSonToFile(File directory, boolean binary) throws IOException {
    // Should we use message order or JMSMessageId?
    File file = new File(directory, getId().substring(3) + ".json");
    if (file.exists())
      throw new IOException("Destination file " + file.getPath() + " already exists.");
    try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
      toJSon(writer, binary);
      writer.flush();
    }
  }
  
  /**
   * Loads a message from transactional persistence.
   * Needed for external tool, to avoid use of AgentServer.getTransaction method.
   * 
   * @param transaction
   * @param txname
   * @return
   * @throws ClassNotFoundException
   * @throws IOException
   */
  public static Message loadMessage(Transaction transaction, String txname) throws ClassNotFoundException, IOException {
    Message msg = (Message) transaction.load(txname);
    if (msg == null)
      throw new IOException("Message not found: " + txname);
    msg.setTxName(txname);
    
    org.objectweb.joram.shared.messages.Message sharedMsg = msg.getSharedMsg();
    //  A message can be marked 'soft' only it this body is not null. 
    if (msg.soft && (sharedMsg.type != sharedMsg.SIMPLE) && (sharedMsg.isNullBody())) {
      msg.msg.setBody(loadBody(transaction, txname), 0, -1);
    }
    return msg;
  }

  // ================================================================================
  // Handling encoding.

  public int getEncodableClassId() {
    return JoramHelper.MESSAGE_CLASS_ID;
  }
  
  /**
   * 
   * Be careful to make the msg.body field null for messages whose body can be swapped (msg.soft).
   */
  public int getEncodedSize() throws Exception {
    int encodedSize = LONG_ENCODED_SIZE + BOOLEAN_ENCODED_SIZE;
    encodedSize += msg.getEncodedSize();
    return encodedSize;
  }

  /**
   * 
   * Be careful to make the msg.body field null for messages whose body can be swapped (msg.soft).
   */
  public void encode(Encoder encoder) throws Exception {
    encoder.encodeUnsignedLong(order);
    // TODO (AF): We could use one byte, and get 8 flags for the same size!!
    encoder.encodeBoolean(soft);
    msg.encode(encoder);
  }

  public void decode(Decoder decoder) throws Exception {
    order = decoder.decodeUnsignedLong();
    // TODO (AF): We could use one byte, and get 8 flags for the same size!!
    soft = decoder.decodeBoolean();
    acksCounter = 0;
    durableAcksCounter = 0;
    msg = new org.objectweb.joram.shared.messages.Message();
    msg.decode(decoder);
  }
  
  public static class MessageFactory implements EncodableFactory {
    public Encodable createEncodable() {
      return new Message();
    }
  }

  // ================================================================================
  // Serializable interface
  
  private void writeObject(ObjectOutputStream out) throws IOException {
    out.writeLong(order);
    out.writeBoolean(soft);

    msg.writeHeaderTo(out);
    if (msg.isFullBody()) {
      StreamUtil.writeTo(msg.getRawBody(), out);
    } else {
      StreamUtil.writeTo(msg.getRawBody(), msg.getBodyOffset(), msg.getBodyLength(), out);
    }
  }

  /**
   * @throws ClassNotFoundException  
   */
  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
    order = in.readLong();
    soft = in.readBoolean();

    acksCounter = 0;
    durableAcksCounter = 0;

    msg = new org.objectweb.joram.shared.messages.Message();
    msg.readHeaderFrom(in);
    msg.setBody(StreamUtil.readByteArrayFrom(in), 0, -1);
  }
}
