1 files changed, 40 insertions, 79 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java
index fa4e85a043..4c9fe81439 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/TransientAMQMessage.java
@@ -44,8 +44,6 @@ public class TransientAMQMessage implements AMQMessage
     /** Used for debugging purposes. */
     protected static final Logger _log = Logger.getLogger(AMQMessage.class);
 
-    private final AtomicInteger _referenceCount = new AtomicInteger(1);
-
     protected ContentHeaderBody _contentHeaderBody;
 
     protected MessagePublishInfo _messagePublishInfo;
@@ -56,23 +54,11 @@ public class TransientAMQMessage implements AMQMessage
 
     protected final Long _messageId;
 
-    /** Flag to indicate that this message requires 'immediate' delivery. */
-
-    private static final byte IMMEDIATE = 0x01;
-
-    /**
-     * Flag to indicate whether this message has been delivered to a consumer. Used in implementing return functionality
-     * for messages published with the 'immediate' flag.
-     */
-
-    private static final byte DELIVERED_TO_CONSUMER = 0x02;
 
     private byte _flags = 0;
 
-    private long _expiration;
-
     private AMQProtocolSession.ProtocolSessionIdentifier _sessionIdentifier;
-    private static final byte IMMEDIATE_AND_DELIVERED = (byte) (IMMEDIATE | DELIVERED_TO_CONSUMER);
+    private long _expiration;
 
     /**
      * Used to iterate through all the body frames associated with this message. Will not keep all the data in memory
@@ -164,14 +150,19 @@ public class TransientAMQMessage implements AMQMessage
 
     public String debugIdentity()
     {
-        return "(HC:" + System.identityHashCode(this) + " ID:" + getMessageId() + " Ref:" + _referenceCount.get() + ")";
+        return "(HC:" + System.identityHashCode(this) + " ID:" + getMessageId() +")";
     }
 
-    public void setExpiration(final long expiration)
+    public void setExpiration(long expiration)
     {
         _expiration = expiration;
     }
 
+    public long getExpiration()
+    {
+        return _expiration;
+    }
+
     public Iterator<AMQDataBlock> getBodyFrameIterator(AMQProtocolSession protocolSession, int channel)
     {
         return new BodyFrameIterator(protocolSession, channel);
@@ -192,57 +183,6 @@ public class TransientAMQMessage implements AMQMessage
         return _messageId;
     }
 
-    /**
-     * Called selectors to determin if the message has already been sent
-     *
-     * @return _deliveredToConsumer
-     */
-    public boolean getDeliveredToConsumer()
-    {
-        return (_flags & DELIVERED_TO_CONSUMER) != 0;
-    }
-
-    /**
-     * Called to enforce the 'immediate' flag.
-     *
-     * @returns true if the message is marked for immediate delivery but has not been marked as delivered
-     * to a consumer
-     */
-    public boolean immediateAndNotDelivered()
-    {
-
-        return (_flags & IMMEDIATE_AND_DELIVERED) == IMMEDIATE;
-
-    }
-
-    /**
-     * Checks to see if the message has expired. If it has the message is dequeued.
-     *
-     * @return true if the message has expire
-     *
-     * @throws AMQException
-     */
-    public boolean expired() throws AMQException
-    {
-
-        if (_expiration != 0L)
-        {
-            long now = System.currentTimeMillis();
-
-            return (now > _expiration);
-        }
-
-        return false;
-    }
-
-    /**
-     * Called when this message is delivered to a consumer. (used to implement the 'immediate' flag functionality).
-     * And for selector efficiency.
-     */
-    public void setDeliveredToConsumer()
-    {
-        _flags |= DELIVERED_TO_CONSUMER;
-    }
 
     public long getSize()
     {
@@ -317,6 +257,11 @@ public class TransientAMQMessage implements AMQMessage
         return false;
     }
 
+    public boolean isImmediate()
+    {
+        return _messagePublishInfo.isImmediate();
+    }
+
     /**
      * This is called when all the content has been received.
      *
@@ -339,33 +284,49 @@ public class TransientAMQMessage implements AMQMessage
             throw new NullPointerException("PublishInfo cannot be null");
         }
 
-        _messagePublishInfo = messagePublishInfo;
+        _arrivalTime = System.currentTimeMillis();
+
+
         _contentHeaderBody = contentHeaderBody;
+        _messagePublishInfo = messagePublishInfo;
 
-        if (contentHeaderBody.bodySize == 0)
-        {
-            _contentBodies = Collections.EMPTY_LIST;
-        }
+        updateHeaderAndFlags();
+    }
 
-        _arrivalTime = System.currentTimeMillis();
+    public long getArrivalTime()
+    {
+        return _arrivalTime;
+    }
 
-        if (messagePublishInfo.isImmediate())
+    public void recoverFromMessageMetaData(MessageMetaData mmd)
+    {
+        _arrivalTime = mmd.getArrivalTime();
+        _contentHeaderBody = mmd.getContentHeaderBody();
+        _messagePublishInfo = mmd.getMessagePublishInfo();
+
+        updateHeaderAndFlags();
+    }
+
+    private void updateHeaderAndFlags()
+    {
+        if (_contentHeaderBody.bodySize == 0)
         {
-            _flags |= IMMEDIATE;
+            _contentBodies = Collections.EMPTY_LIST;
         }
     }
 
-    public long getArrivalTime()
+    public void recoverContentBodyFrame(ContentChunk contentChunk, boolean isLastContentBody) throws AMQException
     {
-        return _arrivalTime;
+        addContentBodyFrame(null, contentChunk, isLastContentBody);
     }
 
+
     public String toString()
     {
         // return "Message[" + debugIdentity() + "]: " + _messageId + "; ref count: " + _referenceCount + "; taken : " +
         // _taken + " by :" + _takenBySubcription;
 
-        return "Message[" + debugIdentity() + "]: " + getMessageId() + "; ref count: " + _referenceCount;
+        return "Message[" + debugIdentity() + "]: " + getMessageId() ;
     }
 
 }