/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.activemq.artemis.core.message.impl;

import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Set;
import java.util.function.Predicate;
import java.util.zip.DataFormatException;
import java.util.zip.Inflater;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.RefCountMessage;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBuffer;
import org.apache.activemq.artemis.core.message.LargeBodyEncoder;
import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.reader.MessageUtil;
import org.apache.activemq.artemis.utils.ByteUtil;
import org.apache.activemq.artemis.utils.DataConstants;
import org.apache.activemq.artemis.utils.UUID;
import org.apache.activemq.artemis.utils.collections.TypedProperties;
import org.jboss.logging.Logger;

/** Note: you shouldn't change properties using multi-threads. Change your properties before you can send it to multiple
 *  consumers */
public class CoreMessage extends RefCountMessage implements ICoreMessage {

   // We use properties to establish routing context on clustering.
   // However if the client resends the message after receiving, it needs to be removed, so we mark these internal
   private static final Predicate<SimpleString> INTERNAL_PROPERTY_NAMES_PREDICATE =
      name -> (name.startsWith(Message.HDR_ROUTE_TO_IDS) && !name.equals(Message.HDR_ROUTE_TO_IDS)) ||
         (name.startsWith(Message.HDR_ROUTE_TO_ACK_IDS) && !name.equals(Message.HDR_ROUTE_TO_ACK_IDS));
   public static final int BUFFER_HEADER_SPACE = PacketImpl.PACKET_HEADERS_SIZE;

   private volatile int memoryEstimate = -1;

   private static final Logger logger = Logger.getLogger(CoreMessage.class);

   // There's an integer with the number of bytes for the body
   public static final int BODY_OFFSET = DataConstants.SIZE_INT;

   /** That is the encode for the whole message, including properties..
       it does not include the buffer for the Packet send and receive header on core protocol */
   protected ByteBuf buffer;

   private volatile boolean validBuffer = false;

   protected volatile ResetLimitWrappedActiveMQBuffer writableBuffer;

   Object body;

   protected int endOfBodyPosition = -1;

   protected int messageIDPosition = -1;

   protected long messageID;

   protected SimpleString address;

   protected byte type;

   protected boolean durable;

   /**
    * GMT milliseconds at which this message expires. 0 means never expires *
    */
   private long expiration;

   protected long timestamp;

   protected byte priority;

   private UUID userID;

   private int propertiesLocation = -1;

   protected volatile TypedProperties properties;

   private final CoreMessageObjectPools coreMessageObjectPools;

   public CoreMessage(final CoreMessageObjectPools coreMessageObjectPools) {
      this.coreMessageObjectPools = coreMessageObjectPools;
   }

   public CoreMessage() {
      this.coreMessageObjectPools = null;
   }

   /** On core there's no delivery annotation */
   @Override
   public Object getAnnotation(SimpleString key) {
      return getObjectProperty(key);
   }

   /** On core there's no delivery annotation */
   @Override
   public Object removeAnnotation(SimpleString key) {
      return removeProperty(key);
   }

   @Override
   public void clearInternalProperties() {
      final TypedProperties properties = this.properties;
      if (properties != null && properties.clearInternalProperties()) {
         messageChanged();
      }
   }

   @Override
   public Persister<Message> getPersister() {
      return CoreMessagePersister.getInstance();
   }

   public CoreMessage initBuffer(final int initialMessageBufferSize) {
      buffer = ActiveMQBuffers.dynamicBuffer(initialMessageBufferSize).byteBuf();

      // There's a bug in netty which means a dynamic buffer won't resize until you write a byte
      buffer.writeByte((byte) 0);

      buffer.setIndex(BODY_OFFSET, BODY_OFFSET);

      return this;
   }

   @Override
   public SimpleString getReplyTo() {
      return getSimpleStringProperty(MessageUtil.REPLYTO_HEADER_NAME);
   }

   @Override
   public RoutingType getRoutingType() {
      if (containsProperty(Message.HDR_ROUTING_TYPE)) {
         return RoutingType.getType(getByteProperty(Message.HDR_ROUTING_TYPE));
      }
      return null;
   }

   @Override
   public Message setRoutingType(RoutingType routingType) {
      if (routingType == null) {
         removeProperty(Message.HDR_ROUTING_TYPE);
      } else {
         putByteProperty(Message.HDR_ROUTING_TYPE, routingType.getType());
      }
      return this;
   }

   @Override
   public CoreMessage setReplyTo(SimpleString address) {

      if (address == null) {
         getProperties().removeProperty(MessageUtil.REPLYTO_HEADER_NAME);
      } else {
         putStringProperty(MessageUtil.REPLYTO_HEADER_NAME, address);
      }
      return this;
   }

   @Override
   public void receiveBuffer(ByteBuf buffer) {
      this.buffer = buffer;
      this.buffer.retain();
      decode(false);
   }

   /** This will fix the incoming body of 1.x messages */
   @Override
   public void receiveBuffer_1X(ByteBuf buffer) {
      this.buffer = buffer;
      this.buffer.retain();
      decode(true);
      validBuffer = false;
   }

   @Override
   public ActiveMQBuffer getReadOnlyBodyBuffer() {
      checkEncode();
      return new ChannelBufferWrapper(buffer.slice(BODY_OFFSET, endOfBodyPosition - BUFFER_HEADER_SPACE).setIndex(0, endOfBodyPosition - BUFFER_HEADER_SPACE).asReadOnly());
   }

   @Override
   public int getBodyBufferSize() {
      checkEncode();
      return endOfBodyPosition - BUFFER_HEADER_SPACE;
   }

   /**
    * This will return the proper buffer to represent the data of the Message. If compressed it will decompress.
    * If large, it will read from the file or streaming.
    * @return
    * @throws ActiveMQException
    */
   @Override
   public ActiveMQBuffer getDataBuffer() {

      ActiveMQBuffer buffer;

      try {
         if (isLargeMessage()) {
            buffer = getLargeMessageBuffer();
         } else {
            buffer = getReadOnlyBodyBuffer();
         }

         if (Boolean.TRUE.equals(getBooleanProperty(Message.HDR_LARGE_COMPRESSED))) {
            buffer = inflate(buffer);
         }
      } catch (Exception e) {
         logger.warn(e.getMessage(), e);
         return getReadOnlyBodyBuffer();
      }

      return buffer;
   }

   private ActiveMQBuffer getLargeMessageBuffer() throws ActiveMQException {
      LargeBodyEncoder encoder = getBodyEncoder();
      encoder.open();
      int bodySize = (int) encoder.getLargeBodySize();
      final ActiveMQBuffer buffer = new ChannelBufferWrapper(UnpooledByteBufAllocator.DEFAULT.heapBuffer(bodySize));
      buffer.byteBuf().ensureWritable(bodySize);
      final ByteBuffer nioBuffer = buffer.byteBuf().internalNioBuffer(0, bodySize);
      encoder.encode(nioBuffer);
      buffer.writerIndex(bodySize);
      encoder.close();
      return buffer;
   }

   private ActiveMQBuffer inflate(ActiveMQBuffer buffer) throws DataFormatException {
      int bytesToRead = buffer.readableBytes();
      Inflater inflater = new Inflater();
      inflater.setInput(ByteUtil.getActiveArray(buffer.readBytes(bytesToRead).toByteBuffer()));

      //get the real size of large message
      long sizeBody = getLongProperty(Message.HDR_LARGE_BODY_SIZE);

      byte[] data = new byte[(int) sizeBody];
      inflater.inflate(data);
      inflater.end();
      ActiveMQBuffer qbuff = ActiveMQBuffers.wrappedBuffer(data);
      qbuff.resetReaderIndex();
      qbuff.resetWriterIndex();
      qbuff.writeBytes(data);
      buffer = qbuff;
      return buffer;
   }

   @Override
   public SimpleString getGroupID() {
      return this.getSimpleStringProperty(Message.HDR_GROUP_ID);
   }

   @Override
   public CoreMessage setGroupID(SimpleString groupId) {
      return this.putStringProperty(Message.HDR_GROUP_ID, groupId);
   }

   @Override
   public CoreMessage setGroupID(String groupId) {
      return this.setGroupID(SimpleString.toSimpleString(groupId, coreMessageObjectPools == null ? null : coreMessageObjectPools.getGroupIdStringSimpleStringPool()));
   }

   @Override
   public int getGroupSequence() {
      return containsProperty(Message.HDR_GROUP_SEQUENCE) ? getIntProperty(Message.HDR_GROUP_SEQUENCE) : 0;
   }

   @Override
   public CoreMessage setGroupSequence(int sequence) {
      return this.putIntProperty(Message.HDR_GROUP_SEQUENCE, sequence);
   }

   @Override
   public Object getCorrelationID() {
      return getObjectProperty(MessageUtil.CORRELATIONID_HEADER_NAME);
   }

   @Override
   public Message setCorrelationID(final Object correlationID) {
      putObjectProperty(MessageUtil.CORRELATIONID_HEADER_NAME, correlationID);
      return this;
   }

   /**
    * @param sendBuffer
    * @param deliveryCount Some protocols (AMQP) will have this as part of the message. ignored on core
    */
   @Override
   public synchronized void sendBuffer(ByteBuf sendBuffer, int deliveryCount) {
      checkEncode();
      sendBuffer.writeBytes(buffer, 0, buffer.writerIndex());
   }

   /**
    * Recast the message as an 1.4 message
    */
   @Override
   public synchronized void sendBuffer_1X(ByteBuf sendBuffer) {
      checkEncode();
      ByteBuf tmpBuffer = buffer.duplicate();
      sendBuffer.writeInt(endOfBodyPosition + DataConstants.SIZE_INT);
      tmpBuffer.readerIndex(DataConstants.SIZE_INT);
      tmpBuffer.readBytes(sendBuffer, endOfBodyPosition - BUFFER_HEADER_SPACE);
      sendBuffer.writeInt(tmpBuffer.writerIndex() + DataConstants.SIZE_INT + BUFFER_HEADER_SPACE);
      tmpBuffer.readBytes(sendBuffer, tmpBuffer.readableBytes());
      sendBuffer.readerIndex(0);
   }

   protected synchronized void checkEncode() {
      if (!validBuffer) {
         encode();
      }
      internalWritableBuffer();
   }

   @Override
   public Long getScheduledDeliveryTime() {
      Object property = getProperties().getProperty(Message.HDR_SCHEDULED_DELIVERY_TIME);

      if (property != null && property instanceof Number) {
         return ((Number) property).longValue();
      }

      return 0L;
   }

   @Override
   public CoreMessage setScheduledDeliveryTime(Long time) {
      if (time == null || time == 0) {
         getProperties().removeProperty(Message.HDR_SCHEDULED_DELIVERY_TIME);
      } else {
         putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, time);
      }
      return this;
   }

   @Override
   public InputStream getBodyInputStream() {
      return null;
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public ActiveMQBuffer getBodyBuffer() {
      // if using the writable buffer, we must parse properties
      getProperties();

      internalWritableBuffer();

      return writableBuffer;
   }

   private void internalWritableBuffer() {
      if (writableBuffer == null) {
         synchronized (this) {
            if (writableBuffer == null) {
               ResetLimitWrappedActiveMQBuffer writableBuffer = new ResetLimitWrappedActiveMQBuffer(BODY_OFFSET, buffer.duplicate(), this);
               if (endOfBodyPosition > 0) {
                  writableBuffer.byteBuf().setIndex(BODY_OFFSET, endOfBodyPosition - BUFFER_HEADER_SPACE + BODY_OFFSET);
                  writableBuffer.resetReaderIndex();
               }
               this.writableBuffer = writableBuffer;
            }
         }
      }
   }

   @Override
   public int getEndOfBodyPosition() {
      if (endOfBodyPosition < 0) {
         endOfBodyPosition = getBodyBuffer().writerIndex();
      }
      return endOfBodyPosition;
   }

   @Override
   public synchronized void messageChanged() {
      //a volatile store is a costly operation: better to check if is necessary
      if (validBuffer) {
         validBuffer = false;
      }
   }

   protected CoreMessage(CoreMessage other) {
      this(other, other.properties);
   }

   public CoreMessage(long id, int bufferSize) {
      this(id, bufferSize, null);
   }

   public CoreMessage(long id, int bufferSize, CoreMessageObjectPools coreMessageObjectPools) {
      this.initBuffer(bufferSize);
      this.setMessageID(id);
      this.coreMessageObjectPools = coreMessageObjectPools;
   }

   protected CoreMessage(CoreMessage other, TypedProperties copyProperties) {
      // This MUST be synchronized using the monitor on the other message to prevent it running concurrently
      // with getEncodedBuffer(), otherwise can introduce race condition when delivering concurrently to
      // many subscriptions and bridging to other nodes in a cluster
      synchronized (other) {
         this.body = other.body;
         this.endOfBodyPosition = other.endOfBodyPosition;
         this.messageID = other.messageID;
         this.address = other.address;
         this.type = other.type;
         this.durable = other.durable;
         this.expiration = other.expiration;
         this.timestamp = other.timestamp;
         this.priority = other.priority;
         this.userID = other.userID;
         this.coreMessageObjectPools = other.coreMessageObjectPools;
         if (copyProperties != null) {
            this.properties = new TypedProperties(copyProperties);
         }
         if (other.buffer != null) {
            this.buffer = other.buffer.copy();
         }
      }
   }

   @Override
   public void copyHeadersAndProperties(final Message msg) {
      messageID = msg.getMessageID();
      address = msg.getAddressSimpleString();
      userID = (UUID) msg.getUserID();
      type = msg.toCore().getType();
      durable = msg.isDurable();
      expiration = msg.getExpiration();
      timestamp = msg.getTimestamp();
      priority = msg.getPriority();

      if (msg instanceof CoreMessage) {
         properties = ((CoreMessage) msg).getProperties();
      }
   }

   @Override
   public Message copy() {
      getProperties();
      checkEncode();
      return new CoreMessage(this);
   }

   @Override
   public Message copy(long newID) {
      return copy().setMessageID(newID);
   }

   @Override
   public long getExpiration() {
      return expiration;
   }

   @Override
   public long getTimestamp() {
      return timestamp;
   }

   @Override
   public CoreMessage setTimestamp(long timestamp) {
      this.timestamp = timestamp;
      return this;
   }

   @Override
   public long getMessageID() {
      return messageID;
   }

   @Override
   public byte getPriority() {
      return priority;
   }

   @Override
   public UUID getUserID() {
      return userID;
   }

   @Override
   public CoreMessage setUserID(Object uuid) {
      this.userID = (UUID) uuid;
      return this;
   }

   @Override
   public String getValidatedUserID() {
      return getStringProperty(Message.HDR_VALIDATED_USER);
   }

   @Override
   public CoreMessage setValidatedUserID(String validatedUserID) {
      putStringProperty(Message.HDR_VALIDATED_USER, value(validatedUserID));
      return this;
   }

   @Override
   public CoreMessage setMessageID(long messageID) {
      this.messageID = messageID;
      if (messageIDPosition >= 0 && validBuffer) {
         buffer.setLong(messageIDPosition, messageID);
      }
      return this;
   }

   @Override
   public CoreMessage setAddress(SimpleString address) {
      if (address == null && this.address == null) {
         // no-op so just return
         return this;
      }
      if (validBuffer && (address == null || !address.equals(this.address))) {
         messageChanged();
      }
      this.address = address;
      return this;
   }

   @Override
   public SimpleString getAddressSimpleString() {
      return address;
   }

   @Override
   public CoreMessage setExpiration(long expiration) {
      this.expiration = expiration;
      messageChanged();
      return this;
   }

   @Override
   public CoreMessage setPriority(byte priority) {
      this.priority = priority;
      messageChanged();
      return this;
   }

   public CoreMessage setUserID(UUID userID) {
      this.userID = userID;
      messageChanged();
      return this;
   }

   /**
    * I am keeping this synchronized as the decode of the Properties is lazy
    */
   public final TypedProperties getProperties() {
      TypedProperties properties = this.properties;
      if (properties == null) {
         properties = getOrInitializeTypedProperties();
      }
      return properties;
   }

   private synchronized TypedProperties getOrInitializeTypedProperties() {
      try {
         TypedProperties properties = this.properties;
         if (properties == null) {
            properties = new TypedProperties(INTERNAL_PROPERTY_NAMES_PREDICATE);
            if (buffer != null && propertiesLocation >= 0) {
               final ByteBuf byteBuf = buffer.duplicate().readerIndex(propertiesLocation);
               properties.decode(byteBuf, coreMessageObjectPools == null ? null : coreMessageObjectPools.getPropertiesDecoderPools());
            }
            this.properties = properties;
         }
         return properties;
      } catch (Throwable e) {
         throw onCheckPropertiesError(e);
      }
   }

   private RuntimeException onCheckPropertiesError(Throwable e) {
      // This is not an expected error, hence no specific logger created
      logger.warn("Could not decode properties for CoreMessage[messageID=" + messageID + ",durable=" + durable + ",userID=" + userID + ",priority=" + priority +
                     ", timestamp=" + timestamp + ",expiration=" + expiration + ",address=" + address + ", propertiesLocation=" + propertiesLocation, e);
      final ByteBuf buffer = this.buffer;
      if (buffer != null) {
         //risky: a racy modification to buffer indexes could break this duplicate operation
         final ByteBuf duplicatebuffer = buffer.duplicate();
         duplicatebuffer.readerIndex(0);
         logger.warn("Failed message has messageID=" + messageID + " and the following buffer:\n" + ByteBufUtil.prettyHexDump(duplicatebuffer));
      } else {
         logger.warn("Failed message has messageID=" + messageID + " and the buffer was null");
      }
      return new RuntimeException(e.getMessage(), e);
   }

   @Override
   public int getMemoryEstimate() {
      if (memoryEstimate == -1) {
         memoryEstimate = memoryOffset +
            (buffer != null ? buffer.capacity() : 0) +
            (properties != null ? properties.getMemoryOffset() : 0);
      }

      return memoryEstimate;
   }

   @Override
   public boolean isServerMessage() {
      // even though CoreMessage is used both on server and client
      // callers are interested in knowing if this is a server large message
      // as it will be used to send the body from the files.
      //
      // this may need further refactoring when we improve large messages
      // and expose that functionality to other protocols.
      return false;
   }

   @Override
   public byte getType() {
      return type;
   }

   @Override
   public CoreMessage setType(byte type) {
      this.type = type;
      return this;
   }

   private void decode(boolean beforeAddress) {
      endOfBodyPosition = buffer.readInt();

      buffer.skipBytes(endOfBodyPosition - BUFFER_HEADER_SPACE);

      decodeHeadersAndProperties(buffer, true);
      buffer.readerIndex(0);
      validBuffer = true;

      if (beforeAddress) {
         endOfBodyPosition = endOfBodyPosition - DataConstants.SIZE_INT;
      }

      internalWritableBuffer();
   }

   public void decodeHeadersAndProperties(final ByteBuf buffer) {
      decodeHeadersAndProperties(buffer, false);
   }

   private void decodeHeadersAndProperties(final ByteBuf buffer, boolean lazyProperties) {
      messageIDPosition = buffer.readerIndex();
      messageID = buffer.readLong();

      address = SimpleString.readNullableSimpleString(buffer, coreMessageObjectPools == null ? null : coreMessageObjectPools.getAddressDecoderPool());
      if (buffer.readByte() == DataConstants.NOT_NULL) {
         byte[] bytes = new byte[16];
         buffer.readBytes(bytes);
         userID = new UUID(UUID.TYPE_TIME_BASED, bytes);
      } else {
         userID = null;
      }
      type = buffer.readByte();
      durable = buffer.readBoolean();
      expiration = buffer.readLong();
      timestamp = buffer.readLong();
      priority = buffer.readByte();
      if (lazyProperties) {
         properties = null;
         propertiesLocation = buffer.readerIndex();
      } else {
         properties = new TypedProperties(INTERNAL_PROPERTY_NAMES_PREDICATE);
         properties.decode(buffer, coreMessageObjectPools == null ? null : coreMessageObjectPools.getPropertiesDecoderPools());
      }
   }

   public synchronized CoreMessage encode() {

      getProperties();

      if (writableBuffer != null) {
         // The message encode takes into consideration the PacketImpl which is not part of this encoding
         // so we always need to take the BUFFER_HEADER_SPACE from packet impl into consideration
         endOfBodyPosition = writableBuffer.writerIndex() + BUFFER_HEADER_SPACE - 4;
      } else if (endOfBodyPosition <= 0) {
         endOfBodyPosition = BUFFER_HEADER_SPACE + DataConstants.SIZE_INT;
      }

      buffer.setIndex(0, 0);
      buffer.writeInt(endOfBodyPosition);

      // The end of body position
      buffer.writerIndex(endOfBodyPosition - BUFFER_HEADER_SPACE + DataConstants.SIZE_INT);

      encodeHeadersAndProperties(buffer);

      validBuffer = true;

      return this;
   }

   public void encodeHeadersAndProperties(final ByteBuf buffer) {
      final TypedProperties properties = getProperties();
      messageIDPosition = buffer.writerIndex();
      buffer.writeLong(messageID);
      SimpleString.writeNullableSimpleString(buffer, address);
      if (userID == null) {
         buffer.writeByte(DataConstants.NULL);
      } else {
         buffer.writeByte(DataConstants.NOT_NULL);
         buffer.writeBytes(userID.asBytes());
      }
      buffer.writeByte(type);
      buffer.writeBoolean(durable);
      buffer.writeLong(expiration);
      buffer.writeLong(timestamp);
      buffer.writeByte(priority);
      properties.encode(buffer);
   }

   @Override
   public int getHeadersAndPropertiesEncodeSize() {
      return DataConstants.SIZE_LONG + // Message ID
         DataConstants.SIZE_BYTE + // user id null?
         (userID == null ? 0 : 16) +
             /* address */SimpleString.sizeofNullableString(address) +
         DataConstants./* Type */SIZE_BYTE +
         DataConstants./* Durable */SIZE_BOOLEAN +
         DataConstants./* Expiration */SIZE_LONG +
         DataConstants./* Timestamp */SIZE_LONG +
         DataConstants./* Priority */SIZE_BYTE +
             /* PropertySize and Properties */getProperties().getEncodeSize();
   }

   @Override
   public Object getDuplicateProperty() {
      return getObjectProperty(Message.HDR_DUPLICATE_DETECTION_ID);
   }

   @Override
   public SimpleString getLastValueProperty() {
      return getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
   }

   @Override
   public Message setLastValueProperty(SimpleString lastValueName) {
      return putStringProperty(Message.HDR_LAST_VALUE_NAME, lastValueName);
   }

   @Override
   public int getEncodeSize() {
      if (buffer == null) {
         return -1;
      }
      checkEncode();
      return buffer.writerIndex();
   }

   @Override
   public boolean isLargeMessage() {
      return false;
   }

   @Override
   public String getAddress() {
      if (address == null) {
         return null;
      } else {
         return address.toString();
      }
   }

   @Override
   public CoreMessage setAddress(String address) {
      messageChanged();
      this.address = SimpleString.toSimpleString(address, coreMessageObjectPools == null ? null : coreMessageObjectPools.getAddressStringSimpleStringPool());
      return this;
   }

   @Override
   public CoreMessage setBuffer(ByteBuf buffer) {
      this.buffer = buffer;

      return this;
   }

   @Override
   public ByteBuf getBuffer() {
      return buffer;
   }

   @Override
   public boolean isDurable() {
      return durable;
   }

   @Override
   public CoreMessage setDurable(boolean durable) {
      messageChanged();
      this.durable = durable;
      return this;
   }

   @Override
   public CoreMessage putBooleanProperty(final String key, final boolean value) {
      return putBooleanProperty(key(key), value);
   }

   @Override
   public CoreMessage putBooleanProperty(final SimpleString key, final boolean value) {
      messageChanged();
      getProperties().putBooleanProperty(key, value);
      return this;
   }

   @Override
   public Boolean getBooleanProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getBooleanProperty(key);
   }

   @Override
   public Boolean getBooleanProperty(final String key) throws ActiveMQPropertyConversionException {
      return getBooleanProperty(key(key));
   }

   @Override
   public CoreMessage putByteProperty(final SimpleString key, final byte value) {
      messageChanged();
      getProperties().putByteProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putByteProperty(final String key, final byte value) {
      return putByteProperty(key(key), value);
   }

   @Override
   public Byte getByteProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getByteProperty(key);
   }

   @Override
   public Byte getByteProperty(final String key) throws ActiveMQPropertyConversionException {
      return getByteProperty(key(key));
   }

   @Override
   public CoreMessage putBytesProperty(final SimpleString key, final byte[] value) {
      messageChanged();
      getProperties().putBytesProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putBytesProperty(final String key, final byte[] value) {
      return putBytesProperty(key(key), value);
   }

   @Override
   public byte[] getBytesProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getBytesProperty(key);
   }

   @Override
   public byte[] getBytesProperty(final String key) throws ActiveMQPropertyConversionException {
      return getBytesProperty(key(key));
   }

   @Override
   public CoreMessage putCharProperty(SimpleString key, char value) {
      messageChanged();
      getProperties().putCharProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putCharProperty(String key, char value) {
      return putCharProperty(key(key), value);
   }

   @Override
   public CoreMessage putShortProperty(final SimpleString key, final short value) {
      messageChanged();
      getProperties().putShortProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putShortProperty(final String key, final short value) {
      return putShortProperty(key(key), value);
   }

   @Override
   public CoreMessage putIntProperty(final SimpleString key, final int value) {
      messageChanged();
      getProperties().putIntProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putIntProperty(final String key, final int value) {
      return putIntProperty(key(key), value);
   }

   @Override
   public Integer getIntProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getIntProperty(key);
   }

   @Override
   public Integer getIntProperty(final String key) throws ActiveMQPropertyConversionException {
      return getIntProperty(key(key));
   }

   @Override
   public CoreMessage putLongProperty(final SimpleString key, final long value) {
      messageChanged();
      getProperties().putLongProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putLongProperty(final String key, final long value) {
      return putLongProperty(key(key), value);
   }

   @Override
   public Long getLongProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getLongProperty(key);
   }

   @Override
   public Long getLongProperty(final String key) throws ActiveMQPropertyConversionException {
      return getLongProperty(key(key));
   }

   @Override
   public CoreMessage putFloatProperty(final SimpleString key, final float value) {
      messageChanged();
      getProperties().putFloatProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putFloatProperty(final String key, final float value) {
      return putFloatProperty(key(key), value);
   }

   @Override
   public CoreMessage putDoubleProperty(final SimpleString key, final double value) {
      messageChanged();
      getProperties().putDoubleProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putDoubleProperty(final String key, final double value) {
      return putDoubleProperty(key(key), value);
   }

   @Override
   public Double getDoubleProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getDoubleProperty(key);
   }

   @Override
   public Double getDoubleProperty(final String key) throws ActiveMQPropertyConversionException {
      return getDoubleProperty(key(key));
   }

   @Override
   public CoreMessage putStringProperty(final SimpleString key, final SimpleString value) {
      messageChanged();
      getProperties().putSimpleStringProperty(key, value);
      return this;
   }

   @Override
   public CoreMessage putStringProperty(final SimpleString key, final String value) {
      return putStringProperty(key, value(value));
   }

   @Override
   public CoreMessage putStringProperty(final String key, final String value) {
      return putStringProperty(key(key), value(value));
   }

   @Override
   public CoreMessage putObjectProperty(final SimpleString key,
                                        final Object value) throws ActiveMQPropertyConversionException {
      messageChanged();
      TypedProperties.setObjectProperty(key, value, getProperties());
      return this;
   }

   @Override
   public Object getObjectProperty(final String key) {
      return getObjectProperty(key(key));
   }

   @Override
   public Object getObjectProperty(final SimpleString key) {
      return getProperties().getProperty(key);
   }

   @Override
   public CoreMessage putObjectProperty(final String key, final Object value) throws ActiveMQPropertyConversionException {
      return putObjectProperty(key(key), value);
   }

   @Override
   public Short getShortProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getShortProperty(key);
   }

   @Override
   public Short getShortProperty(final String key) throws ActiveMQPropertyConversionException {
      return getShortProperty(key(key));
   }

   @Override
   public Float getFloatProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getFloatProperty(key);
   }

   @Override
   public Float getFloatProperty(final String key) throws ActiveMQPropertyConversionException {
      return getFloatProperty(key(key));
   }

   @Override
   public String getStringProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      SimpleString str = getSimpleStringProperty(key);

      if (str == null) {
         return null;
      } else {
         return str.toString();
      }
   }

   @Override
   public String getStringProperty(final String key) throws ActiveMQPropertyConversionException {
      return getStringProperty(key(key));
   }

   @Override
   public SimpleString getSimpleStringProperty(final SimpleString key) throws ActiveMQPropertyConversionException {
      return getProperties().getSimpleStringProperty(key);
   }

   @Override
   public SimpleString getSimpleStringProperty(final String key) throws ActiveMQPropertyConversionException {
      return getSimpleStringProperty(key(key));
   }

   @Override
   public Object removeProperty(final SimpleString key) {
      Object oldValue = getProperties().removeProperty(key);
      if (oldValue != null) {
         messageChanged();
      }
      return oldValue;
   }

   @Override
   public Object removeProperty(final String key) {
      return removeProperty(key(key));
   }

   @Override
   public boolean containsProperty(final SimpleString key) {
      return getProperties().containsProperty(key);
   }

   @Override
   public boolean containsProperty(final String key) {
      return containsProperty(key(key));
   }

   @Override
   public Set<SimpleString> getPropertyNames() {
      return getProperties().getPropertyNames();
   }

   @Override
   public LargeBodyEncoder getBodyEncoder() throws ActiveMQException {
      return new DecodingContext();
   }

   private final class DecodingContext implements LargeBodyEncoder {

      private int lastPos = 0;

      private DecodingContext() {
      }

      @Override
      public void open() {
      }

      @Override
      public void close() {
      }

      @Override
      public long getLargeBodySize() {
         return buffer.writerIndex();
      }

      @Override
      public int encode(final ByteBuffer bufferRead) {
         final int remaining = bufferRead.remaining();
         buffer.getBytes(lastPos, bufferRead);
         lastPos += remaining;
         return remaining;
      }
   }

   @Override
   public int getPersistSize() {
      checkEncode();
      return buffer.writerIndex() + DataConstants.SIZE_INT;
   }

   @Override
   public void persist(ActiveMQBuffer targetRecord) {
      checkEncode();
      targetRecord.writeInt(buffer.writerIndex());
      targetRecord.writeBytes(buffer, 0, buffer.writerIndex());
   }

   @Override
   public void reloadPersistence(ActiveMQBuffer record) {
      int size = record.readInt();
      initBuffer(size);
      buffer.setIndex(0, 0).writeBytes(record.byteBuf(), size);
      decode(false);
   }

   @Override
   public CoreMessage toCore() {
      return this;
   }

   @Override
   public CoreMessage toCore(CoreMessageObjectPools coreMessageObjectPools) {
      return this;
   }

   @Override
   public String toString() {
      try {
         final TypedProperties properties = getProperties();
         return "CoreMessage[messageID=" + messageID + ",durable=" + isDurable() + ",userID=" + getUserID() + ",priority=" + this.getPriority()  +
            ", timestamp=" + toDate(getTimestamp()) + ",expiration=" + toDate(getExpiration()) +
            ", durable=" + durable + ", address=" + getAddress() + ",size=" + getPersistentSize() + ",properties=" + properties + "]@" + System.identityHashCode(this);
      } catch (Throwable e) {
         logger.warn("Error creating String for message: ", e);
         return "ServerMessage[messageID=" + messageID + "]";
      }
   }

   private static String toDate(long timestamp) {
      if (timestamp == 0) {
         return "0";
      } else {
         return new java.util.Date(timestamp).toString();
      }
   }

   private SimpleString key(String key) {
      return SimpleString.toSimpleString(key, getPropertyKeysPool());
   }

   private SimpleString value(String value) {
      return SimpleString.toSimpleString(value, getPropertyValuesPool());
   }

   private SimpleString.StringSimpleStringPool getPropertyKeysPool() {
      return coreMessageObjectPools == null ? null : coreMessageObjectPools.getPropertiesStringSimpleStringPools().getPropertyKeysPool();
   }

   private SimpleString.StringSimpleStringPool getPropertyValuesPool() {
      return coreMessageObjectPools == null ? null : coreMessageObjectPools.getPropertiesStringSimpleStringPools().getPropertyValuesPool();
   }

   @Override
   public long getPersistentSize() throws ActiveMQException {
      return getEncodeSize();
   }
}
