QPID-1630 : Updated MessageFactory to be responsible for assigning messagIDs. Several tests needed updating as they relied upon specified messageIDs.

Added a recovery mode to the MessageFactory where by new messages can have their ids specified but only as long as the new id is larger than the last one.
Recovered messages are always persistent.
Simplified a interfaces as a result of not requiring the messageID to be passed through.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@744140 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 104 insertions, 37 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 3b290b3d51..341aae2875 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -37,7 +37,6 @@ import org.apache.qpid.server.flow.Pre0_10CreditManager;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.IncomingMessage;
-import org.apache.qpid.server.queue.MessageFactory;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.queue.UnauthorizedAccessException;
 import org.apache.qpid.server.queue.AMQMessage;
@@ -108,8 +107,6 @@ public class AMQChannel
 
     private final List<RequiredDeliveryException> _returnMessages = new LinkedList<RequiredDeliveryException>();
 
-    private MessageFactory _messageHandleFactory = new MessageFactory();
-
     // Why do we need this reference ? - ritchiem
     private final AMQProtocolSession _session;
     private boolean _closing; 
@@ -153,8 +150,7 @@ public class AMQChannel
     public void setPublishFrame(MessagePublishInfo info, final Exchange e) throws AMQException
     {
 
-        _currentMessage = new IncomingMessage(_messageStore.getNewMessageId(), info, _txnContext, _session);
-        _currentMessage.setMessageStore(_messageStore);
+        _currentMessage = new IncomingMessage(info, _txnContext, _session, _messageStore);
         _currentMessage.setExchange(e);
     }
 
@@ -178,7 +174,7 @@ public class AMQChannel
 
             routeCurrentMessage();
 
-            _currentMessage.routingComplete(_messageStore, _messageHandleFactory);
+            _currentMessage.routingComplete(_messageStore);
 
             deliverCurrentMessageIfComplete();
 
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
index aad99da6c3..20926b99d7 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
@@ -48,7 +48,6 @@ public class IncomingMessage implements Filterable<RuntimeException>
     private final MessagePublishInfo _messagePublishInfo;
     private ContentHeaderBody _contentHeaderBody;
     private AMQMessage _message;
-    private final Long _messageId;
     private final TransactionalContext _txnContext;
 
     private static final boolean MSG_AUTH = 
@@ -72,22 +71,23 @@ public class IncomingMessage implements Filterable<RuntimeException>
     private long _expiration;
     
     private Exchange _exchange;
+    private static MessageFactory MESSAGE_FACTORY = MessageFactory.getInstance();
 
-    public IncomingMessage(final Long messageId,
-                           final MessagePublishInfo info,
+    public IncomingMessage(final MessagePublishInfo info,
                            final TransactionalContext txnContext,
-                           final AMQProtocolSession publisher)
+                           final AMQProtocolSession publisher,
+                           MessageStore messasgeStore)
     {
-        _messageId = messageId;
         _messagePublishInfo = info;
         _txnContext = txnContext;
         _publisher = publisher;
-
+        _messageStore = messasgeStore;
     }
 
     public void setContentHeaderBody(final ContentHeaderBody contentHeaderBody) throws AMQException
     {
         _contentHeaderBody = contentHeaderBody;
+        _message = MESSAGE_FACTORY.createMessage(_messageStore, isPersistent());
     }
 
     public void setExpiration()
@@ -121,13 +121,10 @@ public class IncomingMessage implements Filterable<RuntimeException>
 
     }
 
-    public void routingComplete(final MessageStore store,
-                                final MessageFactory factory) throws AMQException
+    public void routingComplete(final MessageStore store) throws AMQException
     {
 
-        final boolean persistent = isPersistent();
-        _message = factory.createMessage(_messageId, store, persistent);
-        if (persistent)
+        if (isPersistent())
         {
             _txnContext.beginTranIfNecessary();
              // enqueuing the messages ensure that if required the destinations are recorded to a
@@ -138,7 +135,7 @@ public class IncomingMessage implements Filterable<RuntimeException>
                 for (int i = 0; i < _destinationQueues.size(); i++)
                 {
                     store.enqueueMessage(_txnContext.getStoreContext(),
-                            _destinationQueues.get(i), _messageId);
+                            _destinationQueues.get(i), getMessageId());
                 }
             }
         }
@@ -152,7 +149,7 @@ public class IncomingMessage implements Filterable<RuntimeException>
         // transient message data as quickly as possible
         if (_logger.isDebugEnabled())
         {
-            _logger.debug("Delivering message " + _messageId + " to " + _destinationQueues);
+            _logger.debug("Delivering message " + getMessageId() + " to " + _destinationQueues);
         }
 
         try
@@ -284,14 +281,13 @@ public class IncomingMessage implements Filterable<RuntimeException>
         return false;
     }
 
-    public void setMessageStore(final MessageStore messageStore)
-    {
-        _messageStore = messageStore;
-    }
-
+    /**
+     * The message ID will not be assigned until the ContentHeaderBody has arrived.
+     * @return
+     */
     public Long getMessageId()
     {
-        return _messageId;
+        return _message.getMessageId();
     }
 
     public void setExchange(final Exchange e)
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/MessageFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/MessageFactory.java
index e18834874f..2f6e05963c 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/MessageFactory.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/MessageFactory.java
@@ -21,12 +21,85 @@
 package org.apache.qpid.server.queue;
 
 import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.StoreContext;
+
+import java.util.concurrent.atomic.AtomicLong;
 
 public class MessageFactory
 {
+    private AtomicLong _messageId;
+    private static MessageFactory INSTANCE;
+
+    private enum State
+    {
+        RECOVER,
+        OPEN
+    }
+
+    private State _state = State.RECOVER;
+
+    private MessageFactory()
+    {
+        _messageId = new AtomicLong(0L);
+    }
+
+    public void start()
+    {
+        _state = State.OPEN;
+    }
+
+    /**
+     * Only used by test as test suite is run in a single VM we need to beable to re-enable recovery mode.
+     */    
+    protected void enableRecover()
+    {
+        _state = State.RECOVER;
+    }
+
+    
+    /**
+     * Normal message creation path
+     * @param store
+     * @param persistent
+     * @return
+     */
+    public AMQMessage createMessage(MessageStore store, boolean persistent)
+    {
+        if (_state != State.OPEN)
+        {
+            _state = State.OPEN;
+        }
+
+        return createNextMessage(_messageId.incrementAndGet(), store, persistent);
+    }
+
+    /**
+     * Used for message recovery only and so only creates persistent messages.
+     * @param messageId the id that this message must have
+     * @param store 
+     * @return
+     */
+    public AMQMessage createMessage(Long messageId, MessageStore store)
+    {
+        if (_state != State.RECOVER)
+        {
+            throw new RuntimeException("Unable to create message by ID when not recovering");
+        }
+
+        long currentID = _messageId.get();
+        if (messageId <= currentID)
+        {
+            throw new RuntimeException("Message IDs can only increase current id is:"
+                                       + currentID + ". Requested:" + messageId);
+        }
+        else
+        {
+            _messageId.set(messageId);
+        }
+
+        return createNextMessage(messageId, store, true);
+    }
 
-    public AMQMessage createMessage(Long messageId, MessageStore store, boolean persistent)
+    private AMQMessage createNextMessage(Long messageId, MessageStore store, boolean persistent)
     {
         if (persistent)
         {
@@ -37,5 +110,14 @@ public class MessageFactory
             return new TransientAMQMessage(messageId);
         }
     }
-    
+
+    public static MessageFactory getInstance()
+    {
+        if (INSTANCE == null)
+        {
+            INSTANCE = new MessageFactory();
+        }
+
+        return INSTANCE;
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DerbyMessageStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DerbyMessageStore.java
index 9de2d09b8e..425aed43d4 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DerbyMessageStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/DerbyMessageStore.java
@@ -167,7 +167,7 @@ public class DerbyMessageStore implements MessageStore
 
         // this recovers durable queues and persistent messages
 
-        _messageFactory = new MessageFactory();
+        _messageFactory = MessageFactory.getInstance();
 
         recover();
 
@@ -1355,7 +1355,7 @@ public class DerbyMessageStore implements MessageStore
                 }
                 else
                 {
-                    message = _messageFactory.createMessage(messageId, this, true);                    
+                    message = _messageFactory.createMessage(messageId, this);
 
                     _logger.error("todo must do message recovery now.");
                     //todo must do message recovery now.
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
index f2910acb77..e65dded149 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
@@ -210,13 +210,6 @@ public interface MessageStore
     boolean inTran(StoreContext context);
 
     /**
-     * Return a valid, currently unused message id.
-     *
-     * @return A fresh message id.
-     */
-    Long getNewMessageId();
-
-    /**
      * Stores a chunk of message data.
      *
      * @param context         The transactional context for the operation.