/*
 * 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.shared.messages;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Vector;
import java.util.zip.DataFormatException;
import java.util.zip.Deflater;
import java.util.zip.Inflater;

import org.objectweb.joram.shared.admin.AbstractAdminMessage;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.common.BinaryDump;
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.EncodableHelper;
import fr.dyade.aaa.common.encoding.Encoder;
import fr.dyade.aaa.common.stream.Properties;
import fr.dyade.aaa.common.stream.StreamUtil;
import fr.dyade.aaa.common.stream.Streamable;

/**
 * Implements the <code>Message</code> data structure.
 */
public final class Message implements Cloneable, Serializable, Streamable, Encodable {
  /** define serialVersionUID for interoperability */
  private static final long serialVersionUID = 3L;
  
  /** logger */
  public static final Logger logger = Debug.getLogger(Message.class.getName());

  // default value from  JMS Message
  public static final int NON_PERSISTENT = 1;
  public static final int PERSISTENT = 2;
  public static final int DEFAULT_DELIVERY_MODE = PERSISTENT;
  public static final int DEFAULT_PRIORITY = 4;
  public static final long DEFAULT_TIME_TO_LIVE = 0;
  
  // Reserved property names for Joram
  public static final String SWAPALLOWED = "JMS_JORAM_SWAPALLOWED";
  public static final String ERRORCOUNT = "JMS_JORAM_ERRORCOUNT";
  public static final String ERRORCAUSE_PREFIX = "JMS_JORAM_ERRORCAUSE_";
  public static final String ERRORCODE_PREFIX= "JMS_JORAM_ERRORCODE_";
  public static final String CORRELATION_ID = "JMS_JORAM_CORRELATIONID";
  
  // Reserved property names for MQTTT
  public static final String MQTT_TOPIC = "JMS_JORAM_MQTT_TOPIC";
  public static final String MQTT_QOS = "JMS_JORAM_MQTT_QOS";
  public static final String MQTT_RETAIN = "JMS_JORAM_MQTT_RETAIN";

  // Global configuration properties for the broker.
  public static final String BROKER_USELOADALL = "org.objectweb.joram.mom.messages.USELOADALL";
  public static final String BROKER_SWAPALLOWED = "org.objectweb.joram.mom.messages.SWAPALLOWED";

  /**
   * Constructs a bright new <code>Message</code>.
   */
  public Message() {}

  /** 
   * Byte array containing the body of the message.
   * 
   * The body can be a subarray of this byte array, in this case bodyOffset and bodyLength specify the
   * useful part of the array. If the body completely fills the array then bodyLength value is -1 (and
   * bodyOffset is 0).
   * 
   * If the If body is null, bodyOffset and bodyLength are not significant.
   * 
   * On client side, use getBody and setBody instead of direct access to the body.
   */
  private transient byte[] body = null;
  
  /**
   * The offset of the subarray in <code>body</code> to be used for message body.
   * 
   * /!\ Be careful, this attribute should not be used except in very specific cases. Use getBodyOffset instead.
   * 
   * If body is null, this value is not significant (see getBodyOffset method).
   * 
   * Default value is <code>0</code>.
   */
  private transient int bodyOffset = 0;

  /**
   * Currently unused!!
   * @return
   */
  public final int getBodyOffset() {
    if (isNullBody()) return 0;
    
    // Note: We could always return bodyOffset. 
    if (bodyLength == -1) {
      return 0;
    } else {
      return bodyOffset;
    }
  }
  
  /**
   * The length of the subarray in <code>body</code> to be used for message body.
   * 
   * /!\ Be careful, this attribute should not be used except in very specific cases. Use getBodyLength instead.
   * 
   * Value <code>-1</code> means that the body completely fills the array and <code>bodyOffset</code>
   * can be ignored.
   * 
   * If body is null, this value is not significant (see getBodyLength method).
   * 
   * Default value is <code>-1</code>.
   */
  private transient int bodyLength = -1;
  
  /**
   * Return true if the body fills the entire byte array returned by getRawBody. In this
   * case we don't have to use getBodyOffset and getBodyLength.
   * 
   * @return
   */
  public final boolean isFullBody() {
    return (bodyLength == -1);
  }
  
  /**
   * Gets the number of bytes of the message body.
   * /!\ Do not use broker side, the body could be null (or ensure that the body is loaded).
   *  
   * @return the body length, 0 if body == null.
   */
  public final int getBodyLength() {
    if (isNullBody()) return 0;
    
    if (isFullBody()) {
      return body.length;
    } else {
      return bodyLength;
    }
  }

  /** The message properties table. */
  public transient Properties properties = null;

  /**
   * Returns a property as an object.
   *
   * @param name  The property name.
   */
  public Object getProperty(String name) {
    if (properties == null) return null;
    return properties.get(name);
  }

  /**
   * Sets a property value.
   * If the value is not a Java primitive object (Boolean, Number, String or byte[]) its string representation is used.
   *
   * @param name    The property name.
   * @param value  The property value.
   *
   * @exception IllegalArgumentException  If the key name is illegal (null or empty string).
   */
  public void setProperty(String name, Object value) {
    if (name == null || name.equals(""))
      throw new IllegalArgumentException("Invalid property name: " + name);
    
    if (value == null)
      throw new IllegalArgumentException("Property value cannot be null.");
    
    if (properties == null)
      properties = new Properties();
    
    if (value instanceof Boolean || value instanceof Number || value instanceof String || value instanceof byte[]) {
      properties.put(name, value);
    } else {
      properties.put(name, value.toString());
    }
  }

  /** The message identifier. */
  public transient String id = null;
  
  /** <code>true</code> if the message must be persisted. */
  public transient boolean persistent = true;
 
  /** A simple message carries an empty body. */
  public static final int SIMPLE = 0;
  /** A text message carries a String body. */
  public static final int TEXT = 1;
  /** An object message carries a serializable object. */
  public static final int OBJECT = 2;
  /** A map message carries an hashtable. */
  public static final int MAP = 3;
  /** A stream message carries a bytes stream. */
  public static final int STREAM = 4;
  /** A bytes message carries an array of bytes. */
  public static final int BYTES = 5;
  /** A admin message carries a streamable object. */
  public static final int ADMIN = 6;

  static String[] typeToString = {"SIMPLE", "TEXT", "OBJECT", "MAP", "STREAM", "BYTES", "ADMIN"};
  
  /**
   * The client message type: SIMPLE, TEXT, OBJECT, MAP, STREAM, BYTES, ADMIN.
   * By default, the message type is SIMPLE.
   * Be careful, this type is coded on 4 bits (see writeTo and readFrom methods).
   */
  public transient int type = SIMPLE;

  /**
   * The JMSType header field contains a message type identifier supplied by a
   * client when a message is sent.
   */
  public transient String jmsType = null;

  /**
   * The message priority from 0 to 9, 9 being the highest.
   * By default, the priority is 4.
   * Be careful, this type is coded on 4 bits (see writeTo and readFrom methods).
   */
  public transient int priority = DEFAULT_PRIORITY;
 
  /** The message expiration time, by default 0 for infinite time-to-live. */
  public transient long expiration = 0;

  /** The message time stamp. */
  public transient long timestamp;

  /**
   * <code>true</code> if the message has been denied at least once by a
   * consumer.
   */
  public transient boolean redelivered = false;

  /** The message destination identifier. */
  public transient String toId = null;

  /** The message destination name. */
  public transient String toName = null;

  /** The message destination type. */
  public transient byte toType;
  
  /** <code>true</code> if compressed body. */
  public transient boolean compressed;
  
  /** 
   * If the message body size is greater than the <code>compressedMinSize</code>, this
   * message body is compressed.
   * By default 0, no compression.
   */
  public transient int compressedMinSize = 0;
  
  public transient int compressionLevel = Deflater.BEST_SPEED;
  
  /** the message delivery time value. */
  public transient long deliveryTime;
  
  /** The client connection identification */
  public transient String clientID;

  /**
   * Sets the message destination.
   *
   * @param id  The destination identifier.
   * @param name The destination name.
   * @param type The type of the destination.
   */
  public final void setDestination(String id, String name, byte type) {
    toId = id;
    toName = name;
    toType = type;
  }

//  /** Returns the message destination identifier. */
//  public final String getDestinationId() {
//    return toId;
//  }
//
//  /** Returns <code>true</code> if the destination is a queue. */
//  public final String getDestinationType() {
//    return toType;
//  }

  /** The reply to destination identifier. */
  public transient String replyToId = null;
  
  /** The reply to destination name. */
  public transient String replyToName = null;
  
  /** <code>true</code> if the "reply to" destination is a queue. */
  public transient byte replyToType;

//  /** Returns the destination id the reply should be sent to. */
//  public final String getReplyToId() {
//    return replyToId;
//  }
//
//  /** Returns <code>true</code> if the reply to destination is a queue. */
//  public final String replyToType() {
//    return replyToType;
//  }

  /**
   * Sets the destination to which a reply should be sent.
   *
   * @param id  The destination identifier.
   * @param type The destination type.
   */
  public final void setReplyTo(String id, String name, byte type) {
    replyToId = id;
    replyToName = name;
    replyToType = type;
  }

  /** The correlation identifier field. */
  public transient String correlationId = null;
 
  /**
   * Gets the correlation ID as an array of bytes for the message.
   * 
   * @return the correlation ID for the message as an array of bytes.
   */
  public final byte[] getJMSCorrelationIDAsBytes() {
    if (correlationId != null)
      return correlationId.getBytes();
    return null;
  }
  
  /**
   * Sets the correlation ID as an array of bytes for the message.
   * 
   * @param correlationID the message ID value as an array of bytes. 
   */
  public final void setJMSCorrelationIDAsBytes(byte[] correlationID) {
    if (correlationID != null)
      this.correlationId = new String(correlationID);
    else
      this.correlationId = null;
  }

  /** The number of delivery attempts for this message. */
  public transient  int deliveryCount = 0;

  /**
   * convert serializable object to byte[]
   * 
   * @param object the serializable object
   * @return the byte array
   * @throws IOException In case of error 
   */
  private byte[] toBytes(Serializable object) throws IOException {
  	if (object == null)
  		return null;
  	ByteArrayOutputStream baos = null;
  	ObjectOutputStream oos = null;
  	try {
  		baos = new ByteArrayOutputStream();
  		oos = new ObjectOutputStream(baos);
  		oos.writeObject(object);
  		oos.flush();
  		return baos.toByteArray();
  	} finally {
  		if (oos != null)
  			oos.close();
  		if (baos != null)
  			baos.close();
  	}
  }
  
  /**
   * Convert byte[] to serializable object 
   * 
   * @param body the byte array containing the body.
   * @param offset
   * @param length
   * @return the serializable object
   * @throws Exception In case of error
   */
  private static Serializable fromBytes(byte[] body, int offset, int length) throws Exception {
    if (body == null)  return null;

    Object obj = null;
    try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body, offset, length))) {
      obj = ois.readObject();
    } catch (ClassNotFoundException cnfexc) {
      // Could not build serialized object: reason could be linked to class loaders
      // hierarchy in an application server.
      class Specialized_OIS extends ObjectInputStream {
        Specialized_OIS(InputStream is) throws IOException {
          super(is);
        }

        protected Class resolveClass(ObjectStreamClass osc) throws IOException, ClassNotFoundException {
          String n = osc.getName();
          return Class.forName(n, false, Thread.currentThread().getContextClassLoader());
        }
      }

      try (ObjectInputStream ois = new Specialized_OIS(new ByteArrayInputStream(body, offset, length))) {
        obj = ois.readObject();
      }
    }

    return (Serializable) obj;
  }
  
  // ========================================================================================
  // Methods allowing to get/set message body (used by Mom message
  
  /**
   * Sets a String as the body of the message.
   * @throws IOException In case of an error while setting the text
   */
  public void setText(String text) throws IOException {
  	setBody(toBytes(text));
  }

  /**
   * Returns the text body of the message.
   * Should be used only client side.
   * 
   * @throws Exception In case of an error while getting the text
   */
  public String getText() throws Exception {
    return (String) getObject(body, bodyOffset, bodyLength, compressed);
  }
  
  /**
   * Sets a Map as the body of the message.
   * @throws IOException In case of an error while setting the map
   */
  public void setMap(HashMap map) throws IOException {
    setBody(toBytes(map));
  }
  
  /**
   * Returns the map body of the message.
   * @throws Exception In case of an error while getting the map
   */
  public HashMap getMap() throws Exception {
    return (HashMap) getObject(body, bodyOffset, bodyLength, compressed);
  }

  /**
   * Sets an object as the body of the message. 
   * Should be used only client side, or ensures that the message is completely loaded.
   *
   * @exception IOException  In case of an error while setting the object.
   */
  public void setObject(Serializable object) throws IOException {
    // TODO (AF): Only used in ObjectMessage, type is already set in constructor.
    type = Message.OBJECT;
    setBody(toBytes(object));
  }

  /**
   * Returns the object body of the message.
   *
   * @exception Exception  In case of an error while getting the object.
   */
  public Serializable getObject() throws Exception {
    return getObject(body, bodyOffset, bodyLength, compressed);
  }

  public static Serializable getObject(byte[] body, int offset, int length, boolean compressed) throws Exception {
    if (body == null) return null;
    
    if (compressed) {
      // Note (AF): if the body is compressed then it fills the entire byte array.
      body = uncompress(body);
      length = -1;
      offset = 0;
    }
    
    if (length == -1)
      return fromBytes(body,  0, body.length);
    else
      return fromBytes(body, offset, length);
  }

  // ========================================================================================

  /**
   * Sets an AbstractAdminMessage as the body of the message. 
   *
   * @exception IOException  In case of an error while setting the object.
   */
  public void setAdminMessage(AbstractAdminMessage adminMsg) throws IOException {
    type = Message.ADMIN;

    if (adminMsg == null) {
      clearBody();
    } else {
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      AbstractAdminMessage.write(adminMsg, baos);
      baos.flush();
      setBody(baos.toByteArray(), 0, -1);
      baos.close();
    }
  }

  /**
   * Returns the AbstractAdminMessage body of the message.
   * 
   * TODO (AF): AdminMessage are normally never stored by Queue or Proxy, so body should never be swapped out.
   * We should ensure that these messages are not persistent!
   * 
   * @return the AbstractAdminMessage body of the message.
   */
  public AbstractAdminMessage getAdminMessage() {
    if (isNullBody()) return null;

    ByteArrayInputStream bais = null;
    AbstractAdminMessage adminMsg = null;

    try {
     bais = new ByteArrayInputStream(getBody());
     adminMsg = AbstractAdminMessage.read(bais);
    } catch (Exception e) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR, "getAdminMessage()", e);
      else
        logger.log(BasicLevel.ERROR, "getAdminMessage(): " + e);
    }
    return adminMsg;
  }

  /**
   * Should only be used by MOM message.
   * @return
   */
  public final byte[] getRawBody() {
    if (body == null) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR, "Message body should not be null", new Exception());
      else
        logger.log(BasicLevel.ERROR, "Message body should not be null");
    }
    return body;
  }

  /**
   * Should only be used by MOM message, sets body attribute without any modification of
   * bodyOffset and/or bodyLength.
   * 
   * @param body
   */
  public final void setRawBody(byte[] body) {
    this.body = body;
  }

  /**
   * Sets body, bodyOffset, and bodyLength without any modification.
   * 
   * @param bytes
   * @param bodyOffset
   * @param bodyLength
   */
  public final void setBody(byte[] body, int bodyOffset, int bodyLength) {
    this.body = body;
    this.bodyOffset = bodyOffset;
    this.bodyLength = bodyLength;
  }
  
  /**
   * Set the body.
   * The body is compressed if the body length is greater than compressedMinSize.
   * /!\ Be careful, should be use only client side.
   * 
   * @param bytes a byte array
   * @throws IOException if an I/O error has occurred
   */
  public final void setBody(byte[] bytes) {
    bodyOffset = 0;
    bodyLength = -1;

    if (compressedMinSize > 0 && bytes != null && bytes.length > compressedMinSize) {
      try {
        long initialLength = bytes.length;
        body = compress(bytes, compressionLevel);
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG,
                     "Message compression, type=" + type + ", compressedMinSize=" + compressedMinSize + ", compressionLevel=" + compressionLevel +
                     ", body.length initial=" + initialLength + " final= " + body.length +  ", compression=" + (100-((body.length*100)/initialLength)) + "%");
        compressed = true;
        return;
      } catch (IOException exc) {
        logger.log(BasicLevel.WARN, "Error during body compression: " + exc);
        // If an error occurs during the compression, the body is left unchanged.
      }
    }

    compressed = false;
    body = bytes;
  }
  
  /**
   * Trim the given byte array.
   * 
   * @param bytes   the byte array to trim, should not be null.
   * @param offset
   * @param length
   * @return
   */
  private static byte[] trimBody(byte[] bytes, int offset, int length) {
    if (length >= 0 && (length != bytes.length)) {
      // Create a new byte array that fits the body
      byte[] newBytes = new byte[length];
      System.arraycopy(bytes, offset, newBytes, 0, length);
      return newBytes;
    }
    return bytes;
  }
  
  /**
   *  the body of the message, trims it and uncompress it if needed (the Message body remains unmodified).
   * /!\ Do not use broker side, the body could be null (or ensure that the body is loaded).
   * 
   * @return the body
   * @throws IOException if an I/O error has occurred
   */
  public final byte[] getBody() throws IOException {
    if (body == null) return null;
    
    byte[] bytes = trimBody(body, bodyOffset, bodyLength);
    if (compressed)
      return uncompress(bytes);
    
    return bytes;
  }
  
  /**
   * Set body to null.
   * 
   * /!\ Be careful, only use client side. Broker side, the body can be kept via bodySoftRef mom message attribute.
   */
  public void clearBody() {
    body = null;
    // Useless, non significant values.
    bodyOffset = 0;
    bodyLength = -1;
  }
  
  
  /**
   * Return true if body is null.
   * 
   * /!\ Be careful, only use client side. Broker side, the body can be kept via bodySoftRef mom message attribute.
   * 
   * @return true if body is null
   */
  public final boolean isNullBody() {
    return (body == null);
  }
 
  /**
   * Compress byte array.
   * 
   * @param toCompress a byte array to compress
   * @return the compressed byte array
   * @throws IOException if an I/O error has occurred
   */
  public static byte[] compress(byte[] toCompress, int compressionLevel) throws IOException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "compress(" + toCompress + ')');

    int bodySize = toCompress.length;
    ByteArrayOutputStream baos = new ByteArrayOutputStream(bodySize+4);

    // Writes initial size needed to restore buffer during decompression.
    baos.write((byte) (bodySize >>> 24));
    baos.write((byte) (bodySize >>> 16));
    baos.write((byte) (bodySize >>> 8));
    baos.write((byte) (bodySize >>> 0));

    Deflater compresser = new Deflater(compressionLevel);
    compresser.setInput(toCompress);
    compresser.finish();

    byte[] buff = new byte[1024];
    while(!compresser.finished()) {
      int count = compresser.deflate(buff);
      baos.write(buff, 0, count);
    }
    baos.close();

    compresser.end();
    return baos.toByteArray();
  }
  
  /**
   * Uncompress byte array.
   * 
   * @param toUncompress a compressed byte array
   * @return the uncompressed byte array
   * @throws IOException if an I/O error has occurred
   */
  private static byte[] uncompress(byte[] toUncompress) throws IOException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "uncompress(" + toUncompress + ')');
    
    try {
      Inflater decompresser = new Inflater();
      int size = (((toUncompress[0] &0xFF) << 24) | ((toUncompress[1] &0xFF) << 16) |
          ((toUncompress[2] &0xFF) << 8) | (toUncompress[3] &0xFF));

      byte[] uncompressed = new byte[size];
      decompresser.setInput(toUncompress, 4, toUncompress.length-4);
      int resultLength = decompresser.inflate(uncompressed);
      decompresser.end();
      return uncompressed;
    } catch (DataFormatException e) {
      if (logger.isLoggable(BasicLevel.ERROR))
        logger.log(BasicLevel.ERROR, "", e);
      throw new IOException(e);
    }
  }
  
  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(",id=").append(id);
    strbuf.append(",type=").append(type);
    strbuf.append(",persistent=").append(persistent);
    strbuf.append(",priority=").append(priority);
    strbuf.append(",expiration=").append(expiration);
    strbuf.append(",timestamp=").append(timestamp);
    strbuf.append(",toId=").append(toId);
    if (toName != null)
      strbuf.append('(').append(toName).append(')');
    strbuf.append(",replyToId=").append(replyToId);
    if (replyToName != null)
      strbuf.append('(').append(replyToName).append(')');
    strbuf.append(",correlationId=").append(correlationId);
    strbuf.append(",compressed=").append(compressed);
    strbuf.append(",deliveryTime=").append(deliveryTime);
    strbuf.append(",clientID=").append(clientID);
    strbuf.append(",body=").append(body);
    strbuf.append(')');
  }

  /**
   * Clones the message.
   * /!\ Be careful, if this method is used broker side, we must ensure that the body is loaded.
   */
  public Object clone() {
    try {
      Message clone = (Message) super.clone();
      
      // TODO (AF): May be we can share the body as it should be RO.
      if (body != null) {
        if (isFullBody()) {
          // The message fills the entire byte array.
          clone.body = new byte[body.length];
          System.arraycopy(body, 0, clone.body, 0, body.length);
        } else {
          clone.body = new byte[bodyLength];
          System.arraycopy(body, bodyOffset, clone.body, 0, bodyLength);
        }
        clone.bodyOffset = 0;
        clone.bodyLength = -1;
      }
      // Else if body is null, the cone has already done the work.
      
      if (properties != null) {
        clone.properties = (Properties) properties.clone();
      }
      
      return clone;
    } catch (CloneNotSupportedException cE) {
      // Should never happened!
      return null;
    }
  }

  private static final short typeFlag = 0x0001;
  private static final short replyToIdFlag = 0x0002;
//  private static final short replyToTypeFlag = 0x0004;
  private static final short propertiesFlag = 0x0008;
  private static final short priorityFlag = 0x0010;
  private static final short expirationFlag = 0x0020;
  private static final short corrrelationIdFlag = 0x0040;
  private static final short deliveryCountFlag = 0x0080;
  private static final short jmsTypeFlag = 0x0100;
  private static final short redeliveredFlag = 0x0200;
  private static final short persistentFlag = 0x0400;

  /* ***** ***** ***** ***** *****
   * Streamable interface
   * ***** ***** ***** ***** ***** */

  /**
   *  The object implements the writeTo method to write its contents to the output stream.
   *
   * @param os the stream to write the object to
   */
  public void writeTo(OutputStream os) throws IOException {
    writeHeaderTo(os);
    if (bodyLength == -1) {
      StreamUtil.writeTo(body, os);
    } else {
      StreamUtil.writeTo(body, bodyOffset, bodyLength, os);
    }
  }

  public void writeHeaderTo(OutputStream os) throws IOException {
    StreamUtil.writeTo(id, os);
    StreamUtil.writeTo(toId, os);
    StreamUtil.writeTo(toName, os);
    StreamUtil.writeTo(toType, os);
    StreamUtil.writeTo(timestamp, os);
    StreamUtil.writeTo(compressed, os);
    StreamUtil.writeTo(deliveryTime, os);
    StreamUtil.writeTo(clientID, os);

    // One short is used to know which fields are set
    short s = 0;
    if (type != SIMPLE) { s |= typeFlag; }
    if (replyToId != null) { s |= replyToIdFlag; }
    if (properties != null) { s |= propertiesFlag; }
    if (priority != DEFAULT_PRIORITY) { s |= priorityFlag; }
    if (expiration != 0) { s |= expirationFlag; }
    if (correlationId != null) { s |= corrrelationIdFlag; }
    if (deliveryCount != 0) { s |= deliveryCountFlag; }
    if (jmsType != null) { s |= jmsTypeFlag; }
    if (redelivered) { s |= redeliveredFlag; }
    if (persistent) { s |= persistentFlag; }
    
    StreamUtil.writeTo(s, os);
    
    if (type != SIMPLE) { StreamUtil.writeTo(type, os); }
//    if ((type != SIMPLE) || (priority != DEFAULT_PRIORITY)) {
//      byte b = (byte) (((priority << 4) & 0xF0) | (type & 0x0F));
//      StreamUtil.writeTo(b, os);
//    }
    if (replyToId != null) {
      StreamUtil.writeTo(replyToId, os);
      StreamUtil.writeTo(replyToName, os);
      StreamUtil.writeTo(replyToType, os);
    }
    if (properties != null) { StreamUtil.writeTo(properties, os); }
    if (priority != DEFAULT_PRIORITY) { StreamUtil.writeTo(priority, os); }
    if (expiration != 0) { StreamUtil.writeTo(expiration, os); }
    if (correlationId != null) { StreamUtil.writeTo(correlationId, os); }
    if (deliveryCount != 0) { StreamUtil.writeTo(deliveryCount, os); }
    if (jmsType != null) { StreamUtil.writeTo(jmsType, os); }
  }

  /**
   * The object implements the readFrom method to restore its contents from the input stream.
   * Only used from a new Message (bodyOffset=0, bodyLength=-1).
   *
   * @param is the stream to read data from in order to restore the object
   */
  public void readFrom(InputStream is) throws IOException {
    readHeaderFrom(is);
    body = StreamUtil.readByteArrayFrom(is);
    bodyOffset = 0;
    bodyLength = -1;
  }

  public void readHeaderFrom(InputStream is) throws IOException {
    id = StreamUtil.readStringFrom(is);
    toId = StreamUtil.readStringFrom(is);
    toName = StreamUtil.readStringFrom(is);
    toType = StreamUtil.readByteFrom(is);
    timestamp = StreamUtil.readLongFrom(is);
    compressed = StreamUtil.readBooleanFrom(is);
    deliveryTime = StreamUtil.readLongFrom(is);
    clientID = StreamUtil.readStringFrom(is);
    
    short s = StreamUtil.readShortFrom(is);

    if ((s & typeFlag) != 0) { type = StreamUtil.readIntFrom(is); }
//    if ((s & (typeFlag|priorityFlag)) != 0) {
//      byte b = StreamUtil.readByteFrom(is);
//      type = b & 0x0F;
//      priority = (b >> 4) & 0xF0;
//    } else {
//      type = SIMPLE;
//      priority = DEFAULT_PRIORITY;
//    }
    if ((s & replyToIdFlag) != 0) {
      replyToId = StreamUtil.readStringFrom(is);
      replyToName = StreamUtil.readStringFrom(is);
      replyToType = StreamUtil.readByteFrom(is);
    }
    if ((s & propertiesFlag) != 0) { properties = StreamUtil.readPropertiesFrom(is); }
    priority = DEFAULT_PRIORITY;
    if ((s & priorityFlag) != 0) { priority = StreamUtil.readIntFrom(is); }
    if ((s & expirationFlag) != 0) { expiration = StreamUtil.readLongFrom(is); }
    if ((s & corrrelationIdFlag) != 0) { correlationId = StreamUtil.readStringFrom(is); }
    if ((s & deliveryCountFlag) != 0) { deliveryCount = StreamUtil.readIntFrom(is); }
    if ((s & jmsTypeFlag) != 0) { jmsType = StreamUtil.readStringFrom(is); }
    redelivered = (s & redeliveredFlag) != 0;
    persistent = (s & persistentFlag) != 0;
  }

  /**
   *  this method allows to write to the output stream a vector of message.
   *
   * @param messages 	the vector of messages
   * @param os 		the stream to write the vector to
   */
  public static void writeVectorTo(Vector<Message> messages,
                                   OutputStream os) throws IOException {
    if (messages == null) {
      StreamUtil.writeTo(-1, os);
    } else {
      int size = messages.size();
      StreamUtil.writeTo(size, os);
      for (int i=0; i<size; i++) {
        ((Message) messages.elementAt(i)).writeTo(os);
      }
    }
  }

  /**
   *  this method allows to read from the input stream a vector of messages.
   *
   * @param is 	the stream to read data from in order to restore the vector
   * @return	the vector of messages
   */
  public static Vector<Message> readVectorFrom(InputStream is) throws IOException {
    int size = StreamUtil.readIntFrom(is);
    if (size == -1) return null;

    Vector<Message> messages = new Vector<Message>(size);
    for (int i=0; i<size; i++) {
      Message msg = new Message();
      msg.readFrom(is);
      messages.addElement(msg);
    }
    return messages;
  }

  /** ***** ***** ***** ***** ***** ***** ***** *****
   * Serializable interface
   * ***** ***** ***** ***** ***** ***** ***** ***** */

  private void writeObject(ObjectOutputStream out) throws IOException {
    writeTo(out);
  }

  /**
   * @throws ClassNotFoundException  
   */
  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
    readFrom(in);
  }
  
  /** ***** ***** ***** ***** ***** ***** ***** *****
   * Encodable interface.
   * ***** ***** ***** ***** ***** ***** ***** ***** */
  
  // /!\ Be careful, this code is used to encode/decode MqttMessage in persistence.
  
  public int getEncodableClassId() {
    // Not useful here (no polymorphism).
    return -1;
  }
  
  public int getEncodedSize() throws Exception {
    if (id == null) { // TODO (AF): JORAM-308
      logger.log(BasicLevel.WARN, "Message: Null id.");
      id = "";
    }
    int encodedSize = EncodableHelper.getStringEncodedSize(id);
    
    if (toId == null) { // TODO (AF): JORAM-308
      logger.log(BasicLevel.WARN, "Message: Null toId.");
      toId = "";
    }
    encodedSize += EncodableHelper.getStringEncodedSize(toId);
    encodedSize += EncodableHelper.getNullableStringEncodedSize(toName);
    encodedSize += BYTE_ENCODED_SIZE + LONG_ENCODED_SIZE + BOOLEAN_ENCODED_SIZE + LONG_ENCODED_SIZE + SHORT_ENCODED_SIZE;

    if (type != org.objectweb.joram.shared.messages.Message.SIMPLE) { encodedSize += BYTE_ENCODED_SIZE; }
    if (replyToId != null) {
      encodedSize += EncodableHelper.getStringEncodedSize(replyToId);
      // TODO (AF): Should be a NullableString (be careful to data store compatibility).
      if (replyToName == null) replyToName = "";
      encodedSize += EncodableHelper.getStringEncodedSize(replyToName);
      encodedSize += BYTE_ENCODED_SIZE;
    }
    if (properties != null) { encodedSize += properties.getEncodedSize(); }
    if (priority != org.objectweb.joram.shared.messages.Message.DEFAULT_PRIORITY) { encodedSize += INT_ENCODED_SIZE; }
    if (expiration != 0) { encodedSize += LONG_ENCODED_SIZE; }
    if (correlationId != null) { encodedSize += EncodableHelper.getStringEncodedSize(correlationId); }
    if (deliveryCount != 0) { encodedSize += INT_ENCODED_SIZE; }
    if (jmsType != null) { encodedSize += EncodableHelper.getStringEncodedSize(jmsType); }
    
    if (isFullBody()) {
      // The message fills the entire byte array.
      encodedSize += EncodableHelper.getNullableByteArrayEncodedSize(body);
    } else {
      encodedSize += EncodableHelper.getNullableByteArrayEncodedSize(body, bodyLength);
    }
    encodedSize += EncodableHelper.getNullableStringEncodedSize(clientID);
    
    return encodedSize;
  }
  
  public void encode(Encoder encoder) throws Exception {
    encoder.encodeString(id);
    
    encoder.encodeString(toId);
    encoder.encodeNullableString(toName);
    encoder.encodeByte(toType);
    encoder.encodeUnsignedLong(timestamp);
    encoder.encodeBoolean(compressed);
    encoder.encodeUnsignedLong(deliveryTime);

    // One short is used to know which fields are set
    short s = 0;
    if (type != org.objectweb.joram.shared.messages.Message.SIMPLE) { s |= typeFlag; }
    if (replyToId != null) { s |= replyToIdFlag; }
    if (properties != null) { s |= propertiesFlag; }
    if (priority != org.objectweb.joram.shared.messages.Message.DEFAULT_PRIORITY) { s |= priorityFlag; }
    if (expiration != 0) { s |= expirationFlag; }
    if (correlationId != null) { s |= corrrelationIdFlag; }
    if (deliveryCount != 0) { s |= deliveryCountFlag; }
    if (jmsType != null) { s |= jmsTypeFlag; }
    if (redelivered) { s |= redeliveredFlag; }
    if (persistent) { s |= persistentFlag; }
    
    encoder.encode16(s);
    
    if (type != org.objectweb.joram.shared.messages.Message.SIMPLE) { encoder.encodeByte((byte) type); }
    if (replyToId != null) {
      encoder.encodeString(replyToId);
      // TODO (AF): Should be a NullableString (be careful to data store compatibility).
      encoder.encodeString(replyToName);
      encoder.encodeByte(replyToType);
    }
    if (properties != null) { properties.encode(encoder); }
    if (priority != org.objectweb.joram.shared.messages.Message.DEFAULT_PRIORITY) { encoder.encodeUnsignedInt(priority); }
    if (expiration != 0) { encoder.encodeUnsignedLong(expiration); }
    if (correlationId != null) { encoder.encodeString(correlationId); }
    if (deliveryCount != 0) { encoder.encodeUnsignedInt(deliveryCount); }
    if (jmsType != null) { encoder.encodeString(jmsType); }
    
    if (isFullBody()) {
      // The message fills the entire byte array.
      encoder.encodeNullableByteArray(body);
    } else {
      encoder.encodeNullableByteArray(body, bodyOffset, bodyLength);
    }
    encoder.encodeNullableString(clientID);
  }
  
  public void decode(Decoder decoder) throws Exception {    
    id = decoder.decodeString();
    
    toId = decoder.decodeString();
    toName = decoder.decodeNullableString();
    toType = decoder.decodeByte();
    timestamp = decoder.decodeUnsignedLong();
    compressed = decoder.decodeBoolean();
    deliveryTime = decoder.decodeUnsignedLong();
    
    short s = decoder.decode16();

    if ((s & typeFlag) != 0) { type = decoder.decodeByte(); }
    if ((s & replyToIdFlag) != 0) {
      replyToId = decoder.decodeString();
      replyToName = decoder.decodeString();
      replyToType = decoder.decodeByte();
    }
    if ((s & propertiesFlag) != 0) {
      properties = new Properties();
      properties.decode(decoder); 
    }
    priority = org.objectweb.joram.shared.messages.Message.DEFAULT_PRIORITY;
    if ((s & priorityFlag) != 0) { priority = decoder.decodeUnsignedInt(); }
    if ((s & expirationFlag) != 0) { expiration = decoder.decodeUnsignedLong(); }
    if ((s & corrrelationIdFlag) != 0) { correlationId = decoder.decodeString(); }
    if ((s & deliveryCountFlag) != 0) { deliveryCount = decoder.decodeUnsignedInt(); }
    if ((s & jmsTypeFlag) != 0) { jmsType = decoder.decodeString(); }
    redelivered = (s & redeliveredFlag) != 0;
    persistent = (s & persistentFlag) != 0;
    
    body = decoder.decodeNullableByteArray();
    bodyOffset = 0;
    bodyLength = -1;
    
    clientID = decoder.decodeNullableString();
  }
  
  public static int getMessageVectorEncodedSize(Vector<Message> messages) throws Exception {
    int res = BOOLEAN_ENCODED_SIZE;
    if (messages != null) {
      res += INT_ENCODED_SIZE;
      for (Message msg : messages) {
        res += msg.getEncodedSize();
      }
    }
    return res;
  }
  
  public static void encodeMessageVector(Vector<Message> messages, Encoder encoder) throws Exception {
    if (messages == null) {
      encoder.encodeBoolean(true);
    } else {
      encoder.encodeBoolean(false);
      encoder.encodeUnsignedInt(messages.size());
      for (Message msg : messages) {
        msg.encode(encoder);
      }
    }
  }
  
  public static Vector<Message> decodeMessageVector(Decoder decoder) throws Exception {
    boolean nullFlag = decoder.decodeBoolean();
    if (nullFlag) {
      return null;
    } else {
      int size = decoder.decodeUnsignedInt();
      Vector<Message> messages = new Vector<Message>(size);
      for (int i=0; i<size; i++) {
        Message msg = new Message();
        msg.decode(decoder);
        messages.addElement(msg);
      }
      return messages;
    }
  }

  // ================================================================================
  // Handling export of message in JSon format.
  
//  /**
//   * 
//   * /!\ Be careful, should be used in a client environment (body already loaded).
//   * 
//   * @return
//   * @throws IOException
//   */
//  public String toJSon() throws IOException {
//    StringBuilder strbuf = new StringBuilder();
//    toJSon(strbuf);
//    return strbuf.toString();
//  }
//
//  /**
//   * 
//   * /!\ Be careful, should be used in a client environment (body already loaded).
//   * 
//   * @param appendable
//   * @throws IOException
//   */
//  public void toJSon(Appendable appendable) throws IOException {
//    toJSon(appendable, true);
//  }
//  
//  /**
//   * 
//   * /!\ Be careful, should be used in a client environment (body already loaded).
//   * 
//   * @param appendable
//   * @param binary
//   * @throws IOException
//   */
//  public void toJSon(Appendable appendable, boolean binary) throws IOException {
//    appendable.append("{\n");
//    headerToJSon(appendable);
//    appendable.append(",\n");
//    bodyToJSon(appendable, id, type, body, bodyOffset, bodyLength, compressed, binary);
//    appendable.append("}\n");
//  }

  private void nullableStringToJSon(Appendable appendable, String str) throws IOException {
    if (str == null)
      appendable.append("null");
    else
      appendable.append('"').append(str).append('"');
  }
  
  public void headerToJSon(Appendable appendable) throws IOException {
    appendable.append("\"header\": {\n\"id\": \"").append(id);
    appendable.append("\",\n\"correlationId\": ").append(correlationId);
    
    appendable.append(",\n\"persistent\": ").append(Boolean.toString(persistent));
    appendable.append(",\n\"priority\": ").append(Integer.toString(priority));
    
    appendable.append(",\n\"type\": ").append(Integer.toString(type));
    appendable.append(",\n\"strType\": \"").append(typeToString[type]);
    
    appendable.append("\",\n\"timestamp\": ").append(Long.toString(timestamp));
    appendable.append(",\n\"deliveryTime\": ").append(Long.toString(deliveryTime));
    appendable.append(",\n\"expiration\": ").append(Long.toString(expiration));
    appendable.append(",\n\"redelivered\": ").append(Boolean.toString(redelivered));
    appendable.append(",\n\"deliveryCount\": ").append(Integer.toString(deliveryCount));
    
    appendable.append(",\n\"clientID\": ");
    nullableStringToJSon(appendable, clientID);
    
    appendable.append(",\n\"toId\": ");
    nullableStringToJSon(appendable, toId);
    appendable.append(",\n\"toName\": ");
    nullableStringToJSon(appendable, toName);
    appendable.append(",\n\"toType\": ").append(Byte.toString(toType));
    
    appendable.append(",\n\"replyToId\": ");
    nullableStringToJSon(appendable, replyToId);
    appendable.append(",\n\"replyToName\": ");
    nullableStringToJSon(appendable, replyToName);
    appendable.append(",\n\"replyToType\": ").append(Byte.toString(replyToType));
    
    appendable.append(",\n\"jmsType\": ");
    nullableStringToJSon(appendable, jmsType);
    if (properties == null) {
      appendable.append(",\n\"properties\": null");
    } else {
      appendable.append(",\n\"properties\": ");
      properties.toJSon(appendable);
    }
    
    appendable.append(",\n\"compressed\": ").append(Boolean.toString(compressed));
    // The fields are useless, they are transient and only used client side.
    appendable.append(",\n\"compressedMinSize\": ").append(Integer.toString(compressedMinSize));
    appendable.append(",\n\"compressionLevel\": ").append(Integer.toString(compressionLevel));
    appendable.append("\n}");
  }

  // type -> "SIMPLE", "TEXT", "OBJECT", "MAP", "STREAM", "BYTES", "ADMIN".

  /**
   * 
   * /!\ Be careful, should only be used if body is not null!!
   * 
   * @param appendable
   * @throws IOException
   */
  private static void dumpBodyToJSon(Appendable appendable,
                                     byte[] body, int offset, int length) throws IOException {
    if (length == -1) {
      BinaryDump.toJSon(appendable, body);
    } else {
      BinaryDump.toJSon(appendable, body, offset, length);
    }
  }
  
  /**
   * Exports the specified body to JSon.
   * 
   * Be careful, body must be loaded!!
   * 
   * @param appendable
   * @param msgId The message identifier (used for logging).
   * @param type the type of JMS body.
   * @param body the byte array containing the body.
   * @param offset
   * @param length
   * @param binary
   * @throws IOException
   */
  public static void bodyToJSon(Appendable appendable, String msgId, int type, byte[] body, int offset, int length, boolean compressed, boolean binary) throws IOException {
    if (body == null) {
      appendable.append("\"body\": null");
      return;
    }

    // bodyOffset is not significant, only used part of the byte array is dump.
    appendable.append("\"bodyLength\": ").append(Integer.valueOf(length).toString()).append(",\n");
    appendable.append("\"body\": ");
    dumpBodyToJSon(appendable, body, offset, length);
    
    if (!binary) {
      if (type == 1) { // TextMessage
        String text = null;
        try {
          text = (String) getObject(body, offset, length, compressed);
          appendable.append(",\n\"content\": \"").append(text).append('"');
        } catch (Exception exc) {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.INFO, "Cannot retrieves text from body " + msgId, exc);
          else
            logger.log(BasicLevel.INFO, "Cannot retrieves text from body " + msgId + ": " + exc);
        }
      } else if (type == 2) { // ObjectMessage
        // Be careful, the object class may not be accessible.
        try {
          // try to deserialize the contained object, then prints a String representation.
          Object obj = getObject(body, offset, length, compressed);
          appendable.append(",\n\"content\": \"").append(obj.toString()).append('"');
        } catch (Exception exc) {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.INFO, "Cannot retrieves object from body " + msgId, exc);
          else
            logger.log(BasicLevel.INFO, "Cannot retrieves object from body " + msgId + ": " + exc);
        }
      } else if (type == 3) { // MapMessage
        try {
          HashMap map = (HashMap) getObject(body, offset, length, compressed);
          appendable.append(",\n\"content\": {\n");
          appendable.append("{\n");
          boolean first = true;
          for (Object k : map.keySet()) {
            String key = (String) k;
            Object value = map.get(key);
            if (!first) appendable.append(",\n");
            if (value instanceof String) {
              appendable.append('"').append(key).append("\": \"").append((String) value).append('"');
            } else if (value instanceof byte[]) {
              appendable.append('"').append(key).append("\": ").append(BinaryDump.toJSon((byte[]) value));
            } else {
              appendable.append('"').append(key).append("\": ").append(value.toString());
            }
            first = false;
          }
          appendable.append("\n}");
        } catch (Exception exc) {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.INFO, "Cannot retrieves map from body " + msgId, exc);
          else
            logger.log(BasicLevel.INFO, "Cannot retrieves map from body " + msgId + ": " + exc);
        }
      } else if (type == 4) { // StreamMessage
        // StreamMessage are not self-descriptive, their content cannot be extracted without user help.
      } else if (type == 5) { // BytesMessage
        // BytesMessage are not self-descriptive, their content cannot be extracted without user help.
        try (DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(body))) {
          // try to get a String from the body and prints it.
          String str = inputStream.readUTF();
          appendable.append(",\n\"content\": \"").append(str).append('"');
        } catch (Exception exc) {
          logger.log(BasicLevel.INFO, "Cannot get UTF from BytesMessages " + msgId + ": " + exc);
        }
      } else {
        logger.log(BasicLevel.INFO, "Abnormal type " + type + " for message " + msgId);
      }
    }
    appendable.append('\n');
  }
}
