activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject [09/23] activemq-artemis git commit: ARTEMIS-1009 Pure Message Encoding.
Date Wed, 01 Mar 2017 17:21:03 GMT
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 2943f15..ed9a396 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -43,13 +43,14 @@ import java.util.concurrent.atomic.AtomicLong;
 import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.Message;
+
 import org.apache.activemq.artemis.api.core.Pair;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
 import org.apache.activemq.artemis.api.core.management.ManagementHelper;
 import org.apache.activemq.artemis.core.filter.Filter;
 import org.apache.activemq.artemis.core.io.IOCallback;
-import org.apache.activemq.artemis.core.message.impl.MessageImpl;
+
 import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
 import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
 import org.apache.activemq.artemis.core.persistence.QueueStatus;
@@ -71,7 +72,6 @@ import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.RoutingContext;
 import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.core.server.ScheduledDeliveryHandler;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.server.cluster.RemoteQueueBinding;
 import org.apache.activemq.artemis.core.server.cluster.impl.Redistributor;
 import org.apache.activemq.artemis.core.server.management.ManagementService;
@@ -440,12 +440,12 @@ public class QueueImpl implements Queue {
    }
 
    @Override
-   public void route(final ServerMessage message, final RoutingContext context) throws Exception {
+   public void route(final Message message, final RoutingContext context) throws Exception {
       context.addQueue(address, this);
    }
 
    @Override
-   public void routeWithAck(ServerMessage message, RoutingContext context) {
+   public void routeWithAck(Message message, RoutingContext context) {
       context.addQueueWithAck(address, this);
    }
 
@@ -922,7 +922,7 @@ public class QueueImpl implements Queue {
    }
 
    @Override
-   public boolean hasMatchingConsumer(final ServerMessage message) {
+   public boolean hasMatchingConsumer(final Message message) {
       for (ConsumerHolder holder : consumerList) {
          Consumer consumer = holder.consumer;
 
@@ -1055,7 +1055,7 @@ public class QueueImpl implements Queue {
          pageSubscription.ack((PagedReference) ref);
          postAcknowledge(ref);
       } else {
-         ServerMessage message = ref.getMessage();
+         Message message = ref.getMessage();
 
          boolean durableRef = message.isDurable() && durable;
 
@@ -1087,7 +1087,7 @@ public class QueueImpl implements Queue {
 
          getRefsOperation(tx).addAck(ref);
       } else {
-         ServerMessage message = ref.getMessage();
+         Message message = ref.getMessage();
 
          boolean durableRef = message.isDurable() && durable;
 
@@ -1111,7 +1111,7 @@ public class QueueImpl implements Queue {
 
    @Override
    public void reacknowledge(final Transaction tx, final MessageReference ref) throws Exception {
-      ServerMessage message = ref.getMessage();
+      Message message = ref.getMessage();
 
       if (message.isDurable() && durable) {
          tx.setContainsPersistent();
@@ -1216,11 +1216,11 @@ public class QueueImpl implements Queue {
       return expiryAddress;
    }
 
-   private SimpleString extractAddress(ServerMessage message) {
-      if (message.containsProperty(Message.HDR_ORIG_MESSAGE_ID)) {
-         return message.getSimpleStringProperty(Message.HDR_ORIGINAL_ADDRESS);
+   private SimpleString extractAddress(Message message) {
+      if (message.containsProperty(Message.HDR_ORIG_MESSAGE_ID.toString())) {
+         return message.getSimpleStringProperty(Message.HDR_ORIGINAL_ADDRESS.toString());
       } else {
-         return message.getAddress();
+         return message.getAddressSimpleString();
       }
    }
 
@@ -1244,7 +1244,9 @@ public class QueueImpl implements Queue {
       List<MessageReference> scheduledMessages = scheduledDeliveryHandler.cancel(null);
       if (scheduledMessages != null && scheduledMessages.size() > 0) {
          for (MessageReference ref : scheduledMessages) {
-            ref.getMessage().putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, ref.getScheduledDeliveryTime());
+
+            // TODO-now remove this, use something on Reference
+//            ref.getMessage().putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, ref.getScheduledDeliveryTime());
             ref.setScheduledDeliveryTime(0);
          }
          this.addHead(scheduledMessages, true);
@@ -2274,7 +2276,7 @@ public class QueueImpl implements Queue {
    public boolean checkRedelivery(final MessageReference reference,
                                   final long timeBase,
                                   final boolean ignoreRedeliveryDelay) throws Exception {
-      ServerMessage message = reference.getMessage();
+      Message message = reference.getMessage();
 
       if (internalQueue) {
          if (logger.isTraceEnabled()) {
@@ -2337,7 +2339,7 @@ public class QueueImpl implements Queue {
                      final boolean expiry,
                      final boolean rejectDuplicate,
                      final long... queueIDs) throws Exception {
-      ServerMessage copyMessage = makeCopy(ref, expiry);
+      Message copyMessage = makeCopy(ref, expiry);
 
       copyMessage.setAddress(toAddress);
 
@@ -2346,7 +2348,7 @@ public class QueueImpl implements Queue {
          for (long id : queueIDs) {
             buffer.putLong(id);
          }
-         copyMessage.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, buffer.array());
+         copyMessage.putBytesProperty(Message.HDR_ROUTE_TO_IDS.toString(), buffer.array());
       }
 
       postOffice.route(copyMessage, tx, false, rejectDuplicate);
@@ -2358,16 +2360,17 @@ public class QueueImpl implements Queue {
    private void moveBetweenSnFQueues(final SimpleString queueSuffix,
                                      final Transaction tx,
                                      final MessageReference ref) throws Exception {
-      ServerMessage copyMessage = makeCopy(ref, false, false);
+      Message copyMessage = makeCopy(ref, false, false);
 
       byte[] oldRouteToIDs = null;
       String targetNodeID;
       Binding targetBinding;
 
+      // TODO-now: this needs to go away
       // remove the old route
       for (SimpleString propName : copyMessage.getPropertyNames()) {
-         if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS)) {
-            oldRouteToIDs = (byte[]) copyMessage.removeProperty(propName);
+         if (propName.startsWith(Message.HDR_ROUTE_TO_IDS)) {
+            oldRouteToIDs = (byte[]) copyMessage.removeProperty(propName.toString());
             final String hashcodeToString = oldRouteToIDs.toString(); // don't use Arrays.toString(..) here
             logger.debug("Removed property from message: " + propName + " = " + hashcodeToString + " (" + ByteBuffer.wrap(oldRouteToIDs).getLong() + ")");
 
@@ -2420,7 +2423,7 @@ public class QueueImpl implements Queue {
    }
 
    private Pair<String, Binding> locateTargetBinding(SimpleString queueSuffix,
-                                                     ServerMessage copyMessage,
+                                                     Message copyMessage,
                                                      long oldQueueID) {
       String targetNodeID = null;
       Binding targetBinding = null;
@@ -2440,7 +2443,7 @@ public class QueueImpl implements Queue {
                // parse the queue name of the remote queue binding to determine the node ID
                String temp = remoteQueueBinding.getQueue().getName().toString();
                targetNodeID = temp.substring(temp.lastIndexOf(".") + 1);
-               logger.debug("Message formerly destined for " + oldQueueName + " with ID: " + oldQueueID + " on address " + copyMessage.getAddress() + " on node " + targetNodeID);
+               logger.debug("Message formerly destined for " + oldQueueName + " with ID: " + oldQueueID + " on address " + copyMessage.getAddressSimpleString() + " on node " + targetNodeID);
 
                // now that we have the name of the queue we need to look through all the bindings again to find the new remote queue binding
                for (Map.Entry<SimpleString, Binding> entry2 : postOffice.getAllBindings().entrySet()) {
@@ -2468,14 +2471,14 @@ public class QueueImpl implements Queue {
       return new Pair<>(targetNodeID, targetBinding);
    }
 
-   private ServerMessage makeCopy(final MessageReference ref, final boolean expiry) throws Exception {
+   private Message makeCopy(final MessageReference ref, final boolean expiry) throws Exception {
       return makeCopy(ref, expiry, true);
    }
 
-   private ServerMessage makeCopy(final MessageReference ref,
+   private Message makeCopy(final MessageReference ref,
                                   final boolean expiry,
                                   final boolean copyOriginalHeaders) throws Exception {
-      ServerMessage message = ref.getMessage();
+      Message message = ref.getMessage();
       /*
        We copy the message and send that to the dla/expiry queue - this is
        because otherwise we may end up with a ref with the same message id in the
@@ -2487,7 +2490,15 @@ public class QueueImpl implements Queue {
 
       long newID = storageManager.generateID();
 
-      ServerMessage copy = message.makeCopyForExpiryOrDLA(newID, ref, expiry, copyOriginalHeaders);
+      Message copy = message.copy(newID);
+
+      if (copyOriginalHeaders) {
+         copy.referenceOriginalMessage(message, ref != null ? ref.getQueue().getAddress().toString() : null);
+      }
+
+      if (expiry) {
+         copy.putLongProperty(Message.HDR_ACTUAL_EXPIRY_TIME.toString(), System.currentTimeMillis());
+      }
 
       return copy;
    }
@@ -2549,7 +2560,7 @@ public class QueueImpl implements Queue {
          tx = new TransactionImpl(storageManager);
       }
 
-      ServerMessage copyMessage = makeCopy(ref, reason == AckReason.EXPIRED);
+      Message copyMessage = makeCopy(ref, reason == AckReason.EXPIRED);
 
       copyMessage.setAddress(address);
 
@@ -2719,7 +2730,7 @@ public class QueueImpl implements Queue {
          return;
       }
 
-      ServerMessage message;
+      Message message;
 
       try {
          message = ref.getMessage();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
index 8e3a94b..0f3da07 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
@@ -22,12 +22,12 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.core.paging.cursor.NonExistentPage;
 import org.apache.activemq.artemis.core.persistence.StorageManager;
 import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.Queue;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.apache.activemq.artemis.core.transaction.TransactionOperationAbstract;
 import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
@@ -122,7 +122,7 @@ public class RefsOperation extends TransactionOperationAbstract {
          try {
             Transaction ackedTX = new TransactionImpl(storageManager);
             for (MessageReference ref : ackedRefs) {
-               ServerMessage message = ref.getMessage();
+               Message message = ref.getMessage();
                if (message.isDurable()) {
                   int durableRefCount = message.incrementDurableRefCount();
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
index a5f96b1..4eda3ed 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
@@ -29,6 +29,7 @@ import java.util.Set;
 import java.util.TreeSet;
 
 import org.apache.activemq.artemis.api.core.Message;
+
 import org.apache.activemq.artemis.api.core.Pair;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.client.ClientMessage;
@@ -39,7 +40,7 @@ import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
 import org.apache.activemq.artemis.api.core.management.ManagementHelper;
 import org.apache.activemq.artemis.api.core.management.ResourceNames;
 import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryInternal;
-import org.apache.activemq.artemis.core.message.impl.MessageImpl;
+
 import org.apache.activemq.artemis.core.paging.PagingManager;
 import org.apache.activemq.artemis.core.paging.PagingStore;
 import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
@@ -54,7 +55,6 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.NodeManager;
 import org.apache.activemq.artemis.core.server.Queue;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.server.cluster.ClusterControl;
 import org.apache.activemq.artemis.core.server.cluster.ClusterController;
 import org.apache.activemq.artemis.core.transaction.ResourceManager;
@@ -193,7 +193,7 @@ public class ScaleDownHandler {
                      buffer.putLong(queueID);
                   }
 
-                  message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, buffer.array());
+                  message.putBytesProperty(Message.HDR_ROUTE_TO_IDS.toString(), buffer.array());
 
                   if (logger.isDebugEnabled()) {
                      if (messageReference.isPaged()) {
@@ -203,7 +203,8 @@ public class ScaleDownHandler {
                      }
                   }
 
-                  producer.send(address, message);
+                  // TODO-no: fix encoding
+                  producer.send(address, (Message)message);
                   messageCount++;
 
                   messagesIterator.remove();
@@ -264,11 +265,11 @@ public class ScaleDownHandler {
                byte[] oldRouteToIDs = null;
 
                List<SimpleString> propertiesToRemove = new ArrayList<>();
-               message.removeProperty(MessageImpl.HDR_ROUTE_TO_IDS);
+               message.removeProperty(Message.HDR_ROUTE_TO_IDS.toString());
                for (SimpleString propName : message.getPropertyNames()) {
-                  if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS)) {
+                  if (propName.startsWith(Message.HDR_ROUTE_TO_IDS)) {
                      if (propName.toString().endsWith(propertyEnd)) {
-                        oldRouteToIDs = message.getBytesProperty(propName);
+                        oldRouteToIDs = message.getBytesProperty(propName.toString());
                      }
                      propertiesToRemove.add(propName);
                   }
@@ -277,17 +278,19 @@ public class ScaleDownHandler {
                // TODO: what if oldRouteToIDs == null ??
 
                for (SimpleString propertyToRemove : propertiesToRemove) {
-                  message.removeProperty(propertyToRemove);
+                  message.removeProperty(propertyToRemove.toString());
                }
 
                if (queueOnTarget) {
-                  message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, oldRouteToIDs);
+                  message.putBytesProperty(Message.HDR_ROUTE_TO_IDS.toString(), oldRouteToIDs);
                } else {
-                  message.putBytesProperty(MessageImpl.HDR_SCALEDOWN_TO_IDS, oldRouteToIDs);
+                  message.putBytesProperty(Message.HDR_SCALEDOWN_TO_IDS.toString(), oldRouteToIDs);
                }
 
                logger.debug("Scaling down message " + message + " from " + address + " to " + message.getAddress() + " on node " + targetNodeId);
-               producer.send(message.getAddress(), message);
+
+               // TODO-now: fix encode
+               producer.send(message.getAddress(), (Message)message);
 
                messageCount++;
 
@@ -322,13 +325,13 @@ public class ScaleDownHandler {
          List<TransactionOperation> allOperations = transaction.getAllOperations();
 
          // Get the information of the Prepared TXs so it could replay the TXs
-         Map<ServerMessage, Pair<List<Long>, List<Long>>> queuesToSendTo = new HashMap<>();
+         Map<Message, Pair<List<Long>, List<Long>>> queuesToSendTo = new HashMap<>();
          for (TransactionOperation operation : allOperations) {
             if (operation instanceof PostOfficeImpl.AddOperation) {
                PostOfficeImpl.AddOperation addOperation = (PostOfficeImpl.AddOperation) operation;
                List<MessageReference> refs = addOperation.getRelatedMessageReferences();
                for (MessageReference ref : refs) {
-                  ServerMessage message = ref.getMessage();
+                  Message message = ref.getMessage();
                   Queue queue = ref.getQueue();
                   long queueID;
                   String queueName = queue.getName().toString();
@@ -336,7 +339,7 @@ public class ScaleDownHandler {
                   if (queueIDs.containsKey(queueName)) {
                      queueID = queueIDs.get(queueName);
                   } else {
-                     queueID = createQueueIfNecessaryAndGetID(queueCreateSession, queue, message.getAddress());
+                     queueID = createQueueIfNecessaryAndGetID(queueCreateSession, queue, message.getAddressSimpleString());
                      queueIDs.put(queueName, queueID);  // store it so we don't have to look it up every time
                   }
                   Pair<List<Long>, List<Long>> queueIds = queuesToSendTo.get(message);
@@ -350,7 +353,7 @@ public class ScaleDownHandler {
                RefsOperation refsOperation = (RefsOperation) operation;
                List<MessageReference> refs = refsOperation.getReferencesToAcknowledge();
                for (MessageReference ref : refs) {
-                  ServerMessage message = ref.getMessage();
+                  Message message = ref.getMessage();
                   Queue queue = ref.getQueue();
                   long queueID;
                   String queueName = queue.getName().toString();
@@ -358,7 +361,7 @@ public class ScaleDownHandler {
                   if (queueIDs.containsKey(queueName)) {
                      queueID = queueIDs.get(queueName);
                   } else {
-                     queueID = createQueueIfNecessaryAndGetID(queueCreateSession, queue, message.getAddress());
+                     queueID = createQueueIfNecessaryAndGetID(queueCreateSession, queue, message.getAddressSimpleString());
                      queueIDs.put(queueName, queueID);  // store it so we don't have to look it up every time
                   }
                   Pair<List<Long>, List<Long>> queueIds = queuesToSendTo.get(message);
@@ -373,23 +376,23 @@ public class ScaleDownHandler {
          }
 
          ClientProducer producer = session.createProducer();
-         for (Map.Entry<ServerMessage, Pair<List<Long>, List<Long>>> entry : queuesToSendTo.entrySet()) {
+         for (Map.Entry<Message, Pair<List<Long>, List<Long>>> entry : queuesToSendTo.entrySet()) {
             List<Long> ids = entry.getValue().getA();
             ByteBuffer buffer = ByteBuffer.allocate(ids.size() * 8);
             for (Long id : ids) {
                buffer.putLong(id);
             }
-            ServerMessage message = entry.getKey();
-            message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, buffer.array());
+            Message message = entry.getKey();
+            message.putBytesProperty(Message.HDR_ROUTE_TO_IDS.toString(), buffer.array());
             ids = entry.getValue().getB();
             if (ids.size() > 0) {
                buffer = ByteBuffer.allocate(ids.size() * 8);
                for (Long id : ids) {
                   buffer.putLong(id);
                }
-               message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_ACK_IDS, buffer.array());
+               message.putBytesProperty(Message.HDR_ROUTE_TO_ACK_IDS.toString(), buffer.array());
             }
-            producer.send(message.getAddress(), message);
+            producer.send(message.getAddressSimpleString().toString(), message);
          }
          session.end(xid, XAResource.TMSUCCESS);
          session.prepare(xid);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
index bcc6df1..a130437 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
@@ -31,12 +31,14 @@ 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.ActiveMQIllegalStateException;
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
 import org.apache.activemq.artemis.api.core.management.ManagementHelper;
 import org.apache.activemq.artemis.core.client.impl.ClientConsumerImpl;
 import org.apache.activemq.artemis.core.filter.Filter;
-import org.apache.activemq.artemis.core.message.BodyEncoder;
+import org.apache.activemq.artemis.core.message.LargeBodyEncoder;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
 import org.apache.activemq.artemis.core.persistence.StorageManager;
 import org.apache.activemq.artemis.core.postoffice.Binding;
 import org.apache.activemq.artemis.core.postoffice.QueueBinding;
@@ -48,7 +50,6 @@ import org.apache.activemq.artemis.core.server.LargeServerMessage;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.ServerConsumer;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.server.ServerSession;
 import org.apache.activemq.artemis.core.server.SlowConsumerDetectionListener;
 import org.apache.activemq.artemis.core.server.management.ManagementService;
@@ -205,7 +206,6 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
 
       this.creationTime = System.currentTimeMillis();
 
-
       if (browseOnly) {
          browserDeliverer = new BrowserDeliverer(messageQueue.browserIterator());
       } else {
@@ -341,7 +341,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
             }
             return HandleStatus.BUSY;
          }
-         final ServerMessage message = ref.getMessage();
+         final Message message = ref.getMessage();
 
          if (filter != null && !filter.match(message)) {
             if (logger.isTraceEnabled()) {
@@ -400,7 +400,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
    @Override
    public void proceedDeliver(MessageReference reference) throws Exception {
       try {
-         ServerMessage message = reference.getMessage();
+         Message message = reference.getMessage();
 
          if (message.isLargeMessage() && supportLargeMessage) {
             if (largeMessageDeliverer == null) {
@@ -508,16 +508,13 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
     */
    @Override
    public void forceDelivery(final long sequence) {
-      forceDelivery(sequence, new Runnable() {
-         @Override
-         public void run() {
-            ServerMessage forcedDeliveryMessage = new ServerMessageImpl(storageManager.generateID(), 50);
+      forceDelivery(sequence, () -> {
+         Message forcedDeliveryMessage = new CoreMessage(storageManager.generateID(), 50);
 
-            forcedDeliveryMessage.putLongProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE, sequence);
-            forcedDeliveryMessage.setAddress(messageQueue.getName());
+         forcedDeliveryMessage.putLongProperty(ClientConsumerImpl.FORCED_DELIVERY_MESSAGE, sequence);
+         forcedDeliveryMessage.setAddress(messageQueue.getName());
 
-            callback.sendMessage(null, forcedDeliveryMessage, ServerConsumerImpl.this, 0);
-         }
+         callback.sendMessage(null, forcedDeliveryMessage, ServerConsumerImpl.this, 0);
       });
    }
 
@@ -1018,7 +1015,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
     * @param ref
     * @param message
     */
-   private void deliverStandardMessage(final MessageReference ref, final ServerMessage message) {
+   private void deliverStandardMessage(final MessageReference ref, final Message message) {
       int packetSize = callback.sendMessage(ref, message, ServerConsumerImpl.this, ref.getDeliveryCount());
 
       if (availableCredits != null) {
@@ -1070,7 +1067,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
        */
       private long positionPendingLargeMessage;
 
-      private BodyEncoder context;
+      private LargeBodyEncoder context;
 
       private LargeMessageDeliverer(final LargeServerMessage message, final MessageReference ref) throws Exception {
          largeMessage = message;

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java
deleted file mode 100644
index 39e77ca..0000000
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java
+++ /dev/null
@@ -1,341 +0,0 @@
-/*
- * 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.server.impl;
-
-import java.io.InputStream;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.message.impl.MessageImpl;
-import org.apache.activemq.artemis.core.paging.PagingStore;
-import org.apache.activemq.artemis.core.server.MessageReference;
-import org.apache.activemq.artemis.core.server.Queue;
-import org.apache.activemq.artemis.core.server.ServerMessage;
-import org.apache.activemq.artemis.utils.DataConstants;
-import org.apache.activemq.artemis.utils.MemorySize;
-import org.apache.activemq.artemis.utils.TypedProperties;
-
-public class ServerMessageImpl extends MessageImpl implements ServerMessage {
-
-   private final AtomicInteger durableRefCount = new AtomicInteger();
-
-   private final AtomicInteger refCount = new AtomicInteger();
-
-   private PagingStore pagingStore;
-
-   private static final int memoryOffset;
-
-   private boolean persisted = false;
-
-   static {
-      // This is an estimate of how much memory a ServerMessageImpl takes up, exclusing body and properties
-      // Note, it is only an estimate, it's not possible to be entirely sure with Java
-      // This figure is calculated using the test utilities in org.apache.activemq.tests.unit.util.sizeof
-      // The value is somewhat higher on 64 bit architectures, probably due to different alignment
-
-      if (MemorySize.is64bitArch()) {
-         memoryOffset = 352;
-      } else {
-         memoryOffset = 232;
-      }
-   }
-
-   /*
-    * Constructor for when reading from network
-    */
-   public ServerMessageImpl() {
-   }
-
-   /*
-    * Construct a MessageImpl from storage, or notification, or before routing
-    */
-   public ServerMessageImpl(final long messageID, final int initialMessageBufferSize) {
-      super(initialMessageBufferSize);
-
-      this.messageID = messageID;
-   }
-
-   /*
-    * Copy constructor
-    */
-   protected ServerMessageImpl(final ServerMessageImpl other) {
-      super(other);
-   }
-
-   /*
-    * Copy constructor
-    */
-   protected ServerMessageImpl(final ServerMessageImpl other, TypedProperties properties) {
-      super(other, properties);
-   }
-
-   @Override
-   public boolean isServerMessage() {
-      return true;
-   }
-
-   @Override
-   public ServerMessageImpl setMessageID(final long id) {
-      messageID = id;
-      return this;
-   }
-
-   @Override
-   public MessageReference createReference(final Queue queue) {
-      MessageReference ref = new MessageReferenceImpl(this, queue);
-
-      return ref;
-   }
-
-   @Override
-   public boolean hasInternalProperties() {
-      return properties.hasInternalProperties();
-   }
-
-   @Override
-   public int incrementRefCount() throws Exception {
-      int count = refCount.incrementAndGet();
-
-      if (pagingStore != null) {
-         if (count == 1) {
-            pagingStore.addSize(getMemoryEstimate() + MessageReferenceImpl.getMemoryEstimate());
-         } else {
-            pagingStore.addSize(MessageReferenceImpl.getMemoryEstimate());
-         }
-      }
-
-      return count;
-   }
-
-   @Override
-   public int decrementRefCount() throws Exception {
-      int count = refCount.decrementAndGet();
-
-      if (count < 0) {
-         // this could happen on paged messages since they are not routed and incrementRefCount is never called
-         return count;
-      }
-
-      if (pagingStore != null) {
-         if (count == 0) {
-            pagingStore.addSize(-getMemoryEstimate() - MessageReferenceImpl.getMemoryEstimate());
-
-            if (buffer != null) {
-               // release the buffer now
-               buffer.byteBuf().release();
-            }
-         } else {
-            pagingStore.addSize(-MessageReferenceImpl.getMemoryEstimate());
-         }
-      }
-
-      return count;
-   }
-
-   @Override
-   public int incrementDurableRefCount() {
-      return durableRefCount.incrementAndGet();
-   }
-
-   @Override
-   public int decrementDurableRefCount() {
-      return durableRefCount.decrementAndGet();
-   }
-
-   @Override
-   public int getRefCount() {
-      return refCount.get();
-   }
-
-   @Override
-   public boolean isLargeMessage() {
-      return false;
-   }
-
-   private volatile int memoryEstimate = -1;
-
-   @Override
-   public int getMemoryEstimate() {
-      if (memoryEstimate == -1) {
-         memoryEstimate = ServerMessageImpl.memoryOffset + buffer.capacity() + properties.getMemoryOffset();
-      }
-
-      return memoryEstimate;
-   }
-
-   @Override
-   public ServerMessage copy(final long newID) {
-      ServerMessage m = new ServerMessageImpl(this);
-
-      m.setMessageID(newID);
-
-      return m;
-   }
-
-   @Override
-   public ServerMessage copy() {
-      // This is a simple copy, used only to avoid changing original properties
-      return new ServerMessageImpl(this);
-   }
-
-   public ServerMessage makeCopyForExpiryOrDLA(final long newID,
-                                               MessageReference originalReference,
-                                               final boolean expiry) throws Exception {
-      return makeCopyForExpiryOrDLA(newID, originalReference, expiry, true);
-   }
-
-   @Override
-   public ServerMessage makeCopyForExpiryOrDLA(final long newID,
-                                               MessageReference originalReference,
-                                               final boolean expiry,
-                                               final boolean copyOriginalHeaders) throws Exception {
-      /*
-       We copy the message and send that to the dla/expiry queue - this is
-       because otherwise we may end up with a ref with the same message id in the
-       queue more than once which would barf - this might happen if the same message had been
-       expire from multiple subscriptions of a topic for example
-       We set headers that hold the original message address, expiry time
-       and original message id
-      */
-
-      ServerMessage copy = copy(newID);
-
-      if (copyOriginalHeaders) {
-         copy.setOriginalHeaders(this, originalReference, expiry);
-      }
-
-      return copy;
-   }
-
-   @Override
-   public void setOriginalHeaders(final ServerMessage other,
-                                  final MessageReference originalReference,
-                                  final boolean expiry) {
-      SimpleString originalQueue = other.getSimpleStringProperty(Message.HDR_ORIGINAL_QUEUE);
-
-      if (originalQueue != null) {
-         putStringProperty(Message.HDR_ORIGINAL_QUEUE, originalQueue);
-      } else if (originalReference != null) {
-         putStringProperty(Message.HDR_ORIGINAL_QUEUE, originalReference.getQueue().getName());
-      }
-
-      if (other.containsProperty(Message.HDR_ORIG_MESSAGE_ID)) {
-         putStringProperty(Message.HDR_ORIGINAL_ADDRESS, other.getSimpleStringProperty(Message.HDR_ORIGINAL_ADDRESS));
-
-         putLongProperty(Message.HDR_ORIG_MESSAGE_ID, other.getLongProperty(Message.HDR_ORIG_MESSAGE_ID));
-      } else {
-         putStringProperty(Message.HDR_ORIGINAL_ADDRESS, other.getAddress());
-
-         putLongProperty(Message.HDR_ORIG_MESSAGE_ID, other.getMessageID());
-      }
-
-      // reset expiry
-      setExpiration(0);
-
-      if (expiry) {
-         long actualExpiryTime = System.currentTimeMillis();
-
-         putLongProperty(Message.HDR_ACTUAL_EXPIRY_TIME, actualExpiryTime);
-      }
-
-      bufferValid = false;
-   }
-
-   @Override
-   public void setPagingStore(final PagingStore pagingStore) {
-      this.pagingStore = pagingStore;
-
-      // On the server side, we reset the address to point to the instance of address in the paging store
-      // Otherwise each message would have its own copy of the address String which would take up more memory
-      address = pagingStore.getAddress();
-   }
-
-   @Override
-   public synchronized void forceAddress(final SimpleString address) {
-      this.address = address;
-      bufferValid = false;
-   }
-
-   @Override
-   public PagingStore getPagingStore() {
-      return pagingStore;
-   }
-
-   @Override
-   public boolean storeIsPaging() {
-      if (pagingStore != null) {
-         return pagingStore.isPaging();
-      } else {
-         return false;
-      }
-   }
-
-   @Override
-   public String toString() {
-      try {
-         return "ServerMessage[messageID=" + messageID + ",durable=" + isDurable() + ",userID=" + getUserID() + ",priority=" + this.getPriority() + ", bodySize=" + this.getBodyBufferDuplicate().capacity() +
-            ", timestamp=" + toDate(getTimestamp()) + ",expiration=" + toDate(getExpiration()) +
-            ", durable=" + durable + ", address=" + getAddress() + ",properties=" + properties.toString() + "]@" + System.identityHashCode(this);
-      } catch (Throwable e) {
-         return "ServerMessage[messageID=" + messageID + "]";
-      }
-   }
-
-   private static String toDate(long timestamp) {
-      if (timestamp == 0) {
-         return "0";
-      } else {
-         return new java.util.Date(timestamp).toString();
-      }
-
-   }
-
-   @Override
-   public InputStream getBodyInputStream() {
-      return null;
-   }
-
-   // Encoding stuff
-
-   @Override
-   public void encodeMessageIDToBuffer() {
-      // We first set the message id - this needs to be set on the buffer since this buffer will be re-used
-
-      buffer.setLong(buffer.getInt(MessageImpl.BUFFER_HEADER_SPACE) + DataConstants.SIZE_INT, messageID);
-   }
-
-   @Override
-   public byte[] getDuplicateIDBytes() {
-      Object duplicateID = getDuplicateProperty();
-
-      if (duplicateID == null) {
-         return null;
-      } else {
-         if (duplicateID instanceof SimpleString) {
-            return ((SimpleString) duplicateID).getData();
-         } else {
-            return (byte[]) duplicateID;
-         }
-      }
-   }
-
-   @Override
-   public Object getDuplicateProperty() {
-      return getObjectProperty(Message.HDR_DUPLICATE_DETECTION_ID);
-   }
-}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index 52ecda1..27fbdcb 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -31,6 +31,7 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.activemq.artemis.Closeable;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ActiveMQIOErrorException;
 import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
@@ -46,7 +47,6 @@ import org.apache.activemq.artemis.core.exception.ActiveMQXAException;
 import org.apache.activemq.artemis.core.filter.Filter;
 import org.apache.activemq.artemis.core.filter.impl.FilterImpl;
 import org.apache.activemq.artemis.core.io.IOCallback;
-import org.apache.activemq.artemis.core.message.impl.MessageInternal;
 import org.apache.activemq.artemis.core.paging.PagingManager;
 import org.apache.activemq.artemis.core.paging.PagingStore;
 import org.apache.activemq.artemis.core.persistence.OperationContext;
@@ -66,14 +66,13 @@ import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
 import org.apache.activemq.artemis.core.server.AddressQueryResult;
 import org.apache.activemq.artemis.core.server.BindingQueryResult;
-import org.apache.activemq.artemis.core.server.LargeServerMessage;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.QueueQueryResult;
 import org.apache.activemq.artemis.core.server.RoutingContext;
 import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.core.server.ServerConsumer;
-import org.apache.activemq.artemis.core.server.ServerMessage;
+
 import org.apache.activemq.artemis.core.server.ServerSession;
 import org.apache.activemq.artemis.core.server.TempQueueObserver;
 import org.apache.activemq.artemis.core.server.management.ManagementService;
@@ -155,9 +154,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
 
    private final SimpleString managementAddress;
 
-   // The current currentLargeMessage being processed
-   private volatile LargeServerMessage currentLargeMessage;
-
    protected final RoutingContext routingContext = new RoutingContextImpl(null);
 
    protected final SessionCallback callback;
@@ -187,6 +183,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
 
    private Map<SimpleString, RoutingType> prefixes;
 
+   private Set<Closeable> closeables;
+
    public ServerSessionImpl(final String name,
                             final String username,
                             final String password,
@@ -273,6 +271,14 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
    }
 
    @Override
+   public void addCloseable(Closeable closeable) {
+      if (closeables == null) {
+         closeables = new HashSet<>();
+      }
+      this.closeables.add(closeable);
+   }
+
+   @Override
    public void disableSecurity() {
       this.securityEnabled = false;
    }
@@ -376,11 +382,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
 
       consumers.clear();
 
-      if (currentLargeMessage != null) {
-         try {
-            currentLargeMessage.deleteFile();
-         } catch (Throwable error) {
-            ActiveMQServerLogger.LOGGER.errorDeletingLargeMessageFile(error);
+      if (closeables != null) {
+         for (Closeable closeable : closeables) {
+            closeable.close(failed);
          }
       }
 
@@ -1272,30 +1276,12 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
    }
 
    @Override
-   public void sendLarge(final MessageInternal message) throws Exception {
-      // need to create the LargeMessage before continue
-      long id = storageManager.generateID();
-
-      LargeServerMessage largeMsg = storageManager.createLargeMessage(id, message);
-
-      if (logger.isTraceEnabled()) {
-         logger.trace("sendLarge::" + largeMsg);
-      }
-
-      if (currentLargeMessage != null) {
-         ActiveMQServerLogger.LOGGER.replacingIncompleteLargeMessage(currentLargeMessage.getMessageID());
-      }
-
-      currentLargeMessage = largeMsg;
-   }
-
-   @Override
-   public RoutingStatus send(final ServerMessage message, final boolean direct) throws Exception {
+   public RoutingStatus send(final Message message, final boolean direct) throws Exception {
       return send(message, direct, false);
    }
 
    @Override
-   public RoutingStatus send(final ServerMessage message,
+   public RoutingStatus send(final Message message,
                              final boolean direct,
                              boolean noAutoCreateQueue) throws Exception {
       return send(getCurrentTransaction(), message, direct, noAutoCreateQueue);
@@ -1303,7 +1289,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
 
    @Override
    public RoutingStatus send(Transaction tx,
-                             final ServerMessage message,
+                             final Message message,
                              final boolean direct,
                              boolean noAutoCreateQueue) throws Exception {
 
@@ -1319,19 +1305,18 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       //case the id header already generated.
       if (!message.isLargeMessage()) {
          long id = storageManager.generateID();
-
+         // This will re-encode the message
          message.setMessageID(id);
-         message.encodeMessageIDToBuffer();
       }
 
       if (server.getConfiguration().isPopulateValidatedUser() && validatedUser != null) {
          message.putStringProperty(Message.HDR_VALIDATED_USER, SimpleString.toSimpleString(validatedUser));
       }
 
-      SimpleString address = removePrefix(message.getAddress());
+      SimpleString address = removePrefix(message.getAddressSimpleString());
 
       // In case the prefix was removed, we also need to update the message
-      if (address != message.getAddress()) {
+      if (address != message.getAddressSimpleString()) {
          message.setAddress(address);
       }
 
@@ -1340,14 +1325,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       }
 
       if (address == null) {
-         if (message.isDurable()) {
-            // We need to force a re-encode when the message gets persisted or when it gets reloaded
-            // it will have no address
-            message.setAddress(defaultAddress);
-         } else {
-            // We don't want to force a re-encode when the message gets sent to the consumer
-            message.setAddressTransient(defaultAddress);
-         }
+         // We don't want to force a re-encode when the message gets sent to the consumer
+         message.setAddress(defaultAddress);
       }
 
       if (logger.isTraceEnabled()) {
@@ -1359,7 +1338,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
          throw ActiveMQMessageBundle.BUNDLE.noAddress();
       }
 
-      if (message.getAddress().equals(managementAddress)) {
+      if (message.getAddressSimpleString().equals(managementAddress)) {
          // It's a management message
 
          handleManagementMessage(tx, message, direct);
@@ -1369,32 +1348,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       return result;
    }
 
-   @Override
-   public void sendContinuations(final int packetSize,
-                                 final long messageBodySize,
-                                 final byte[] body,
-                                 final boolean continues) throws Exception {
-      if (currentLargeMessage == null) {
-         throw ActiveMQMessageBundle.BUNDLE.largeMessageNotInitialised();
-      }
-
-      // Immediately release the credits for the continuations- these don't contribute to the in-memory size
-      // of the message
-
-      currentLargeMessage.addBytes(body);
-
-      if (!continues) {
-         currentLargeMessage.releaseResources();
-
-         if (messageBodySize >= 0) {
-            currentLargeMessage.putLongProperty(Message.HDR_LARGE_BODY_SIZE, messageBodySize);
-         }
-
-         doSend(tx, currentLargeMessage, false, false);
-
-         currentLargeMessage = null;
-      }
-   }
 
    @Override
    public void requestProducerCredits(SimpleString address, final int credits) throws Exception {
@@ -1566,10 +1519,6 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       connectionFailed(me, failedOver);
    }
 
-   public void clearLargeMessage() {
-      currentLargeMessage = null;
-   }
-
    private void installJMSHooks() {
    }
 
@@ -1588,10 +1537,10 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
    }
 
    private RoutingStatus handleManagementMessage(final Transaction tx,
-                                                 final ServerMessage message,
+                                                 final Message message,
                                                  final boolean direct) throws Exception {
       try {
-         securityCheck(removePrefix(message.getAddress()), CheckType.MANAGE, this);
+         securityCheck(removePrefix(message.getAddressSimpleString()), CheckType.MANAGE, this);
       } catch (ActiveMQException e) {
          if (!autoCommitSends) {
             tx.markAsRollbackOnly(e);
@@ -1599,7 +1548,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
          throw e;
       }
 
-      ServerMessage reply = managementService.handleMessage(message);
+      Message reply = managementService.handleMessage(message);
 
       SimpleString replyTo = message.getSimpleStringProperty(ClientMessageImpl.REPLYTO_HEADER_NAME);
 
@@ -1669,8 +1618,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       theTx.rollback();
    }
 
+   @Override
    public RoutingStatus doSend(final Transaction tx,
-                               final ServerMessage msg,
+                               final Message msg,
                                final boolean direct,
                                final boolean noAutoCreateQueue) throws Exception {
       RoutingStatus result = RoutingStatus.OK;
@@ -1683,7 +1633,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       if (msg.containsProperty(Message.HDR_ROUTING_TYPE)) {
          routingType = RoutingType.getType(msg.getByteProperty(Message.HDR_ROUTING_TYPE));
       }
-      Pair<SimpleString, RoutingType> art = getAddressAndRoutingType(msg.getAddress(), routingType);
+      Pair<SimpleString, RoutingType> art = getAddressAndRoutingType(msg.getAddressSimpleString(), routingType);
 
       // Consumer
       // check the user has write access to this address.
@@ -1707,12 +1657,14 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
 
          result = postOffice.route(msg, routingContext, direct);
 
-         Pair<UUID, AtomicLong> value = targetAddressInfos.get(msg.getAddress());
+         Pair<UUID, AtomicLong> value = targetAddressInfos.get(msg.getAddressSimpleString());
 
          if (value == null) {
-            targetAddressInfos.put(msg.getAddress(), new Pair<>(msg.getUserID(), new AtomicLong(1)));
+            // TODO-now: userID
+            targetAddressInfos.put(msg.getAddressSimpleString(), new Pair<>((UUID)msg.getUserID(), new AtomicLong(1)));
          } else {
-            value.setA(msg.getUserID());
+            // TODO-now: userID
+            value.setA((UUID)msg.getUserID());
             value.getB().incrementAndGet();
          }
       } finally {

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/ManagementService.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/ManagementService.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/ManagementService.java
index 0222928..84ab636 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/ManagementService.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/ManagementService.java
@@ -21,6 +21,7 @@ import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
 
 import org.apache.activemq.artemis.api.core.BroadcastGroupConfiguration;
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.TransportConfiguration;
 import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder;
@@ -42,7 +43,7 @@ import org.apache.activemq.artemis.core.server.Divert;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.QueueFactory;
 import org.apache.activemq.artemis.api.core.RoutingType;
-import org.apache.activemq.artemis.core.server.ServerMessage;
+
 import org.apache.activemq.artemis.core.server.cluster.Bridge;
 import org.apache.activemq.artemis.core.server.cluster.BroadcastGroup;
 import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
@@ -128,5 +129,5 @@ public interface ManagementService extends NotificationService, ActiveMQComponen
 
    Object[] getResources(Class<?> resourceType);
 
-   ServerMessage handleMessage(ServerMessage message) throws Exception;
+   Message handleMessage(Message message) throws Exception;
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
index 55f2aea..002b2c7 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
@@ -34,6 +34,8 @@ import java.util.concurrent.ScheduledExecutorService;
 
 import org.apache.activemq.artemis.api.core.BroadcastGroupConfiguration;
 import org.apache.activemq.artemis.api.core.JsonUtil;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.TransportConfiguration;
 import org.apache.activemq.artemis.api.core.management.AcceptorControl;
@@ -56,6 +58,7 @@ import org.apache.activemq.artemis.core.management.impl.BroadcastGroupControlImp
 import org.apache.activemq.artemis.core.management.impl.ClusterConnectionControlImpl;
 import org.apache.activemq.artemis.core.management.impl.DivertControlImpl;
 import org.apache.activemq.artemis.core.management.impl.QueueControlImpl;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
 import org.apache.activemq.artemis.core.messagecounter.MessageCounter;
 import org.apache.activemq.artemis.core.messagecounter.MessageCounterManager;
 import org.apache.activemq.artemis.core.messagecounter.impl.MessageCounterManagerImpl;
@@ -71,13 +74,10 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
 import org.apache.activemq.artemis.core.server.Divert;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.QueueFactory;
-import org.apache.activemq.artemis.api.core.RoutingType;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.server.cluster.Bridge;
 import org.apache.activemq.artemis.core.server.cluster.BroadcastGroup;
 import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
 import org.apache.activemq.artemis.core.server.impl.AddressInfo;
-import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
 import org.apache.activemq.artemis.core.server.management.ManagementService;
 import org.apache.activemq.artemis.core.server.management.Notification;
 import org.apache.activemq.artemis.core.server.management.NotificationListener;
@@ -365,9 +365,9 @@ public class ManagementServiceImpl implements ManagementService {
    }
 
    @Override
-   public ServerMessage handleMessage(final ServerMessage message) throws Exception {
+   public Message handleMessage(final Message message) throws Exception {
       // a reply message is sent with the result stored in the message body.
-      ServerMessage reply = new ServerMessageImpl(storageManager.generateID(), 512);
+      Message reply = new CoreMessage(storageManager.generateID(), 512);
 
       String resourceName = message.getStringProperty(ManagementHelper.HDR_RESOURCE_NAME);
       if (logger.isDebugEnabled()) {
@@ -631,7 +631,7 @@ public class ManagementServiceImpl implements ManagementService {
 
                long messageID = storageManager.generateID();
 
-               ServerMessage notificationMessage = new ServerMessageImpl(messageID, 512);
+               Message notificationMessage = new CoreMessage(messageID, 512);
 
                // Notification messages are always durable so the user can choose whether to add a durable queue to
                // consume them in

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionDetail.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionDetail.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionDetail.java
index efe4cf9..e686a2c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionDetail.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/TransactionDetail.java
@@ -26,8 +26,9 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.activemq.artemis.api.core.JsonUtil;
+
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.core.server.MessageReference;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.transaction.impl.XidImpl;
 import org.apache.activemq.artemis.utils.JsonLoader;
 
@@ -97,7 +98,7 @@ public abstract class TransactionDetail {
 
             msgJson.add(KEY_MSG_OP_TYPE, opType);
 
-            ServerMessage msg = ref.getMessage().copy();
+            Message msg = ref.getMessage().copy();
 
             msgJson.add(KEY_MSG_TYPE, decodeMessageType(msg));
             JsonUtil.addToObject(KEY_MSG_PROPERTIES, decodeMessageProperties(msg), msgJson);
@@ -108,7 +109,7 @@ public abstract class TransactionDetail {
       return detailJson.build();
    }
 
-   public abstract String decodeMessageType(ServerMessage msg);
+   public abstract String decodeMessageType(Message msg);
 
-   public abstract Map<String, Object> decodeMessageProperties(ServerMessage msg);
+   public abstract Map<String, Object> decodeMessageProperties(Message msg);
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/CoreTransactionDetail.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/CoreTransactionDetail.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/CoreTransactionDetail.java
index 4730596..c885341 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/CoreTransactionDetail.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/transaction/impl/CoreTransactionDetail.java
@@ -20,7 +20,7 @@ import javax.transaction.xa.Xid;
 import java.util.Map;
 
 import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.core.server.ServerMessage;
+
 import org.apache.activemq.artemis.core.transaction.Transaction;
 import org.apache.activemq.artemis.core.transaction.TransactionDetail;
 
@@ -31,7 +31,7 @@ public class CoreTransactionDetail extends TransactionDetail {
    }
 
    @Override
-   public String decodeMessageType(ServerMessage msg) {
+   public String decodeMessageType(Message msg) {
       int type = msg.getType();
       switch (type) {
          case Message.DEFAULT_TYPE: // 0
@@ -52,7 +52,7 @@ public class CoreTransactionDetail extends TransactionDetail {
    }
 
    @Override
-   public Map<String, Object> decodeMessageProperties(ServerMessage msg) {
+   public Map<String, Object> decodeMessageProperties(Message msg) {
       return msg.toMap();
    }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessageConverter.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessageConverter.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessageConverter.java
index a342e13..3a5e2bf 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessageConverter.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessageConverter.java
@@ -16,12 +16,11 @@
  */
 package org.apache.activemq.artemis.spi.core.protocol;
 
-import org.apache.activemq.artemis.core.server.ServerMessage;
+import org.apache.activemq.artemis.api.core.Message;
 
-// TODO: use this interface properly on OpenWire
 public interface MessageConverter {
 
-   ServerMessage inbound(Object messageInbound) throws Exception;
+   Message inbound(Object messageInbound) throws Exception;
 
-   Object outbound(ServerMessage messageOutbound, int deliveryCount) throws Exception;
+   Object outbound(Message messageOutbound, int deliveryCount) throws Exception;
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java
new file mode 100644
index 0000000..14891f5
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java
@@ -0,0 +1,88 @@
+/**
+ * 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.spi.core.protocol;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.jboss.logging.Logger;
+
+public class MessagePersister implements Persister<Message> {
+
+   private static final Logger logger = Logger.getLogger(MessagePersister.class);
+
+   private static final MessagePersister theInstance = new MessagePersister();
+
+   /** This will be used for reading messages */
+   private static Map<Byte, Persister<Message>> protocols = new ConcurrentHashMap<>();
+
+
+   public static void registerProtocol(ProtocolManagerFactory manager) {
+      Persister<Message> messagePersister = manager.getPersister();
+      if (messagePersister == null) {
+         logger.warn("Cannot find persister for " + manager);
+      } else {
+         registerPersister(manager.getStoreID(), manager.getPersister());
+      }
+   }
+
+   public static void clearPersisters() {
+      protocols.clear();
+   }
+
+   public static void registerPersister(byte recordType, Persister<Message> persister) {
+      protocols.put(recordType, persister);
+   }
+
+   public static MessagePersister getInstance() {
+      return theInstance;
+   }
+
+
+   protected MessagePersister() {
+   }
+
+   protected byte getID() {
+      return (byte)0;
+   }
+
+   @Override
+   public int getEncodeSize(Message record) {
+      return 0;
+   }
+
+
+   /** Sub classes must add the first short as the protocol-id */
+   @Override
+   public void encode(ActiveMQBuffer buffer, Message record) {
+      buffer.writeByte(getID());
+   }
+
+   @Override
+   public Message decode(ActiveMQBuffer buffer, Message record) {
+      byte protocol = buffer.readByte();
+      Persister<Message> persister = protocols.get(protocol);
+      if (persister == null) {
+         throw new NullPointerException("couldn't find factory for type=" + protocol);
+      }
+      return persister.decode(buffer, record);
+   }
+}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java
index 890fbfe..c2b7334 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java
@@ -28,6 +28,8 @@ import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 
+/**
+ * Info: ProtocolManager is loaded by {@link org.apache.activemq.artemis.core.remoting.server.impl.RemotingServiceImpl#loadProtocolManagerFactories(Iterable)} */
 public interface ProtocolManager<P extends BaseInterceptor> {
 
    ProtocolManagerFactory<P> getFactory();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
index d3b1b2e..9574540 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
@@ -20,10 +20,25 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.activemq.artemis.api.core.BaseInterceptor;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.persistence.Persister;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 
 public interface ProtocolManagerFactory<P extends BaseInterceptor> {
 
+   /** This is to be used to store the protocol-id on Messages.
+    *  Messages are stored on their bare format.
+    *  The protocol manager will be responsible to code or decode messages.
+    *  The caveat here is that the first short-sized bytes need to be this constant. */
+   default byte getStoreID() {
+      return (byte)0;
+   }
+
+   default Persister<Message> getPersister() {
+      return null;
+   }
+
+
    /**
     * When you create the ProtocolManager, you should filter out any interceptors that won't belong
     * to this Protocol.

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/SessionCallback.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/SessionCallback.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/SessionCallback.java
index ee236c7..799e8b0 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/SessionCallback.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/SessionCallback.java
@@ -16,10 +16,10 @@
  */
 package org.apache.activemq.artemis.spi.core.protocol;
 
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.ServerConsumer;
-import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
 
 public interface SessionCallback {
@@ -55,10 +55,10 @@ public interface SessionCallback {
    //       and I wanted to avoid re-fetching paged data in case of GCs on this specific case.
    //
    //       Future developments may change this, but beware why I have chosen to keep the parameter separated here
-   int sendMessage(MessageReference ref, ServerMessage message, ServerConsumer consumerID, int deliveryCount);
+   int sendMessage(MessageReference ref, Message message, ServerConsumer consumerID, int deliveryCount);
 
    int sendLargeMessage(MessageReference reference,
-                        ServerMessage message,
+                        Message message,
                         ServerConsumer consumerID,
                         long bodySize,
                         int deliveryCount);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/main/resources/schema/artemis-configuration.xsd
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
index 0c33a35..6fdef44 100644
--- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd
+++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -682,22 +682,6 @@
             </xsd:annotation>
          </xsd:element>
 
-         <xsd:element name="perf-blast-pages" type="xsd:int" default="-1" maxOccurs="1" minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  XXX Only meant to be used by project developers
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
-         <xsd:element name="run-sync-speed-test" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
-            <xsd:annotation>
-               <xsd:documentation>
-                  XXX Only meant to be used by project developers
-               </xsd:documentation>
-            </xsd:annotation>
-         </xsd:element>
-
          <xsd:element name="server-dump-interval" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
index f374979..5e9a95a 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
@@ -77,7 +77,6 @@ public class ConfigurationImplTest extends ActiveMQTestBase {
       Assert.assertEquals(ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, conf.getJournalBufferSize_AIO());
       Assert.assertEquals(ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, conf.getJournalBufferSize_NIO());
       Assert.assertEquals(ActiveMQDefaultConfiguration.isDefaultJournalLogWriteRate(), conf.isLogJournalWriteRate());
-      Assert.assertEquals(ActiveMQDefaultConfiguration.getDefaultJournalPerfBlastPages(), conf.getJournalPerfBlastPages());
       Assert.assertEquals(ActiveMQDefaultConfiguration.isDefaultMessageCounterEnabled(), conf.isMessageCounterEnabled());
       Assert.assertEquals(ActiveMQDefaultConfiguration.getDefaultMessageCounterMaxDayHistory(), conf.getMessageCounterMaxDayHistory());
       Assert.assertEquals(ActiveMQDefaultConfiguration.getDefaultMessageCounterSamplePeriod(), conf.getMessageCounterSamplePeriod());
@@ -232,10 +231,6 @@ public class ConfigurationImplTest extends ActiveMQTestBase {
          conf.setLogJournalWriteRate(b);
          Assert.assertEquals(b, conf.isLogJournalWriteRate());
 
-         i = RandomUtil.randomInt();
-         conf.setJournalPerfBlastPages(i);
-         Assert.assertEquals(i, conf.getJournalPerfBlastPages());
-
          l = RandomUtil.randomLong();
          conf.setServerDumpInterval(l);
          Assert.assertEquals(l, conf.getServerDumpInterval());
@@ -434,10 +429,6 @@ public class ConfigurationImplTest extends ActiveMQTestBase {
       conf.setLogJournalWriteRate(b);
       Assert.assertEquals(b, conf.isLogJournalWriteRate());
 
-      i = RandomUtil.randomInt();
-      conf.setJournalPerfBlastPages(i);
-      Assert.assertEquals(i, conf.getJournalPerfBlastPages());
-
       l = RandomUtil.randomLong();
       conf.setServerDumpInterval(l);
       Assert.assertEquals(l, conf.getServerDumpInterval());

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/test/java/org/apache/activemq/artemis/core/filter/impl/FilterTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/filter/impl/FilterTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/filter/impl/FilterTest.java
index d73accd..1eb749b 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/filter/impl/FilterTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/filter/impl/FilterTest.java
@@ -18,10 +18,10 @@ package org.apache.activemq.artemis.core.filter.impl;
 
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ActiveMQInvalidFilterExpressionException;
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.filter.Filter;
-import org.apache.activemq.artemis.core.server.ServerMessage;
-import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
 import org.apache.activemq.artemis.tests.util.SilentTestCase;
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.junit.Assert;
@@ -35,13 +35,13 @@ public class FilterTest extends SilentTestCase {
 
    private Filter filter;
 
-   private ServerMessage message;
+   private Message message;
 
    @Override
    @Before
    public void setUp() throws Exception {
       super.setUp();
-      message = new ServerMessageImpl(1, 1000);
+      message = new CoreMessage().initBuffer(1024).setMessageID(1);
    }
 
    @Test
@@ -59,7 +59,7 @@ public class FilterTest extends SilentTestCase {
 
       message.putStringProperty(new SimpleString("color"), new SimpleString("RED"));
       Assert.assertTrue(filter.match(message));
-      message = new ServerMessageImpl();
+      message = new CoreMessage();
       Assert.assertFalse(filter.match(message));
    }
 
@@ -94,7 +94,7 @@ public class FilterTest extends SilentTestCase {
 
       filter = FilterImpl.createFilter(new SimpleString("AMQDurable='NON_DURABLE'"));
 
-      message = new ServerMessageImpl();
+      message = new CoreMessage();
       message.setDurable(true);
 
       Assert.assertFalse(filter.match(message));

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/53887656/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/group/impl/ClusteredResetMockTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/group/impl/ClusteredResetMockTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/group/impl/ClusteredResetMockTest.java
index 0e9a3f2..92204ce 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/group/impl/ClusteredResetMockTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/group/impl/ClusteredResetMockTest.java
@@ -23,6 +23,7 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.artemis.api.core.BroadcastGroupConfiguration;
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.TransportConfiguration;
 import org.apache.activemq.artemis.api.core.management.ManagementHelper;
@@ -44,7 +45,7 @@ import org.apache.activemq.artemis.core.server.Divert;
 import org.apache.activemq.artemis.core.server.Queue;
 import org.apache.activemq.artemis.core.server.QueueFactory;
 import org.apache.activemq.artemis.api.core.RoutingType;
-import org.apache.activemq.artemis.core.server.ServerMessage;
+
 import org.apache.activemq.artemis.core.server.cluster.Bridge;
 import org.apache.activemq.artemis.core.server.cluster.BroadcastGroup;
 import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
@@ -329,7 +330,7 @@ public class ClusteredResetMockTest extends ActiveMQTestBase {
       }
 
       @Override
-      public ServerMessage handleMessage(ServerMessage message) throws Exception {
+      public Message handleMessage(Message message) throws Exception {
          return null;
       }
 


Mime
View raw message