Fixes to tests.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/new_persistence@480107 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 122 insertions, 74 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 338e557d9e..f725fe2871 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -29,6 +29,7 @@ import org.apache.qpid.server.ack.UnacknowledgedMessage;
 import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
 import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
 import org.apache.qpid.server.exchange.MessageRouter;
+import org.apache.qpid.server.exchange.NoRouteException;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.queue.AMQMessage;
 import org.apache.qpid.server.queue.AMQQueue;
@@ -110,7 +111,7 @@ public class AMQChannel
         _prefetch_LowWaterMark = _prefetch_HighWaterMark / 2;
         _messageStore = messageStore;
         _exchanges = exchanges;
-        // TODO: fix me to pass in the txn context or at least have a factory for it
+        // by default the session is non-transactional
         _txnContext = new NonTransactionalContext(_messageStore, this, _returnMessages);
     }
 
@@ -119,7 +120,7 @@ public class AMQChannel
      */
     public void setLocalTransactional()
     {
-        _txnContext = new LocalTransactionalContext(_messageStore, new TxnBuffer(_messageStore), _returnMessages);
+        _txnContext = new LocalTransactionalContext(_messageStore, new TxnBuffer(), _returnMessages);
     }
 
     public boolean isTransactional()
@@ -168,9 +169,10 @@ public class AMQChannel
 
     public void setPublishFrame(BasicPublishBody publishBody, AMQProtocolSession publisher) throws AMQException
     {
-        _currentMessage = new AMQMessage(_messageStore.getNewMessageId(), publishBody, _txnContext);
+        _currentMessage = new AMQMessage(_messageStore.getNewMessageId(), publishBody,
+                                         _txnContext);
         // TODO: used in clustering only I think (RG)
-        //_currentMessage.setPublisher(publisher);
+        _currentMessage.setPublisher(publisher);
     }
 
     public void publishContentHeader(ContentHeaderBody contentHeaderBody)
@@ -194,7 +196,7 @@ public class AMQChannel
         }
     }
 
-    public void publishContentBody(ContentBody contentBody)
+    public void publishContentBody(ContentBody contentBody, AMQProtocolSession protocolSession)
             throws AMQException
     {
         if (_currentMessage == null)
@@ -208,6 +210,9 @@ public class AMQChannel
         {
             if (_currentMessage.addContentBodyFrame(contentBody))
             {
+                // callback to allow the context to do any post message processing
+                // primary use is to allow message return processing in the non-tx case
+                _txnContext.messageProcessed(protocolSession);
                 _currentMessage = null;
             }
         }
@@ -222,7 +227,14 @@ public class AMQChannel
 
     protected void routeCurrentMessage() throws AMQException
     {
-        _exchanges.routeContent(_currentMessage);
+        try
+        {
+            _exchanges.routeContent(_currentMessage);
+        }
+        catch (NoRouteException e)
+        {
+            _returnMessages.add(e);
+        }
     }
 
     /*protected void routeCurrentMessage2() throws AMQException
diff --git a/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java b/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java
index d8253156a0..b85e3603b7 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java
@@ -7,9 +7,9 @@
  * 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
@@ -36,6 +36,7 @@ public abstract class RequiredDeliveryException extends AMQException
     {
         super(message);
         _amqMessage = payload;
+        payload.incrementReference();
     }
 
     public AMQMessage getAMQMessage()
diff --git a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java
index f39ea73268..b7a75d5b71 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java
@@ -7,9 +7,9 @@
  * 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
@@ -20,11 +20,12 @@
  */
 package org.apache.qpid.server.ack;
 
-import org.apache.qpid.server.txn.TransactionalContext;
 import org.apache.qpid.AMQException;
+import org.apache.qpid.server.txn.TransactionalContext;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Set;
 
 public interface UnacknowledgedMessageMap
 {
@@ -61,5 +62,13 @@ public interface UnacknowledgedMessageMap
     int size();
 
     void clear();
+
+    UnacknowledgedMessage get(long deliveryTag);
+
+    /**
+     * Get the set of delivery tags that are outstanding.
+     * @return a set of delivery tags
+     */
+    Set<Long> getDeliveryTags();
 }
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java
index 05121ff1ab..1f4333549a 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java
@@ -98,6 +98,7 @@ public class UnacknowledgedMessageMapImpl implements UnacknowledgedMessageMap
             {
                 visitor.callback(msg);
             }
+            visitor.visitComplete();
         }
     }
 
@@ -186,7 +187,8 @@ public class UnacknowledgedMessageMapImpl implements UnacknowledgedMessageMap
             }
         }
     }
-    private UnacknowledgedMessage get(long key)
+
+    public UnacknowledgedMessage get(long key)
     {
         synchronized (_lock)
         {
@@ -194,6 +196,14 @@ public class UnacknowledgedMessageMapImpl implements UnacknowledgedMessageMap
         }
     }
 
+    public Set<Long> getDeliveryTags()
+    {
+        synchronized (_lock)
+        {
+            return _map.keySet();
+        }
+    }
+
     private void collect(long key, List<UnacknowledgedMessage> msgs)
     {
         synchronized (_lock)
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
index 7294f6b374..46289ecab8 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
@@ -368,7 +368,8 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
             ProtocolInitiation pi = (ProtocolInitiation) message;
             // this ensures the codec never checks for a PI message again
             ((AMQDecoder)_codecFactory.getDecoder()).setExpectProtocolInitiation(false);
-            try {
+            try
+            {
                 pi.checkVersion(this); // Fails if not correct
                 // This sets the protocol version (and hence framing classes) for this session.
                 _major = pi.protocolMajor;
@@ -378,7 +379,9 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
                 AMQFrame response = ConnectionStartBody.createAMQFrame((short)0, pi.protocolMajor, pi.protocolMinor, null,
                                                                        mechanisms.getBytes(), locales.getBytes());
                 _minaProtocolSession.write(response);
-            } catch (AMQException e) {
+            }
+            catch (AMQException e)
+            {
                 _logger.error("Received incorrect protocol initiation", e);
                 /* Find last protocol version in protocol version list. Make sure last protocol version
                 listed in the build file (build-module.xml) is the latest version which will be used
@@ -474,7 +477,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
         {
             _logger.debug("Content body frame received: " + frame);
         }
-        getChannel(frame.channel).publishContentBody((ContentBody)frame.bodyFrame);
+        getChannel(frame.channel).publishContentBody((ContentBody)frame.bodyFrame, this);
     }
 
     /**
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
index d7e317cfa5..73264c5310 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
@@ -44,8 +44,7 @@ public class AMQMessage
     private final Set<Object> _tokens = new HashSet<Object>();
 
     /**
-     * Used in clustering
-     * TODO need to get rid of this
+     * Only use in clustering - should ideally be removed?
      */
     private AMQProtocolSession _publisher;
 
@@ -156,7 +155,8 @@ public class AMQMessage
         }
     }
 
-    public AMQMessage(long messageId, BasicPublishBody publishBody, TransactionalContext txnContext)
+    public AMQMessage(long messageId, BasicPublishBody publishBody,
+                      TransactionalContext txnContext)
     {
         _messageId = messageId;
         _txnContext = txnContext;
@@ -175,16 +175,41 @@ public class AMQMessage
      * @param txnContext
      * @param contentHeader
      */
-    public AMQMessage(long messageId, BasicPublishBody publishBody, TransactionalContext txnContext,
-                      ContentHeaderBody contentHeader) throws AMQException
+    public AMQMessage(long messageId, BasicPublishBody publishBody,
+                      TransactionalContext txnContext, ContentHeaderBody contentHeader) throws AMQException
     {
         this(messageId, publishBody, txnContext);
         setContentHeaderBody(contentHeader);
     }
 
+    /**
+     * Used in testing only. This allows the passing of the content header and some body fragments on
+     * construction.
+     * @param messageId
+     * @param publishBody
+     * @param txnContext
+     * @param contentHeader
+     * @param destinationQueues
+     * @param contentBodies
+     * @throws AMQException
+     */
+    public AMQMessage(long messageId, BasicPublishBody publishBody,
+                      TransactionalContext txnContext,
+                      ContentHeaderBody contentHeader, List<AMQQueue> destinationQueues,
+                      List<ContentBody> contentBodies, MessageStore messageStore,
+                      MessageHandleFactory messageHandleFactory) throws AMQException
+    {
+        this(messageId, publishBody, txnContext, contentHeader);
+        _destinationQueues = destinationQueues;
+        routingComplete(messageStore, messageHandleFactory);
+        for (ContentBody cb : contentBodies)
+        {
+            addContentBodyFrame(cb);
+        }
+    }
+
     protected AMQMessage(AMQMessage msg) throws AMQException
     {
-        _publisher = msg._publisher;
         _messageId = msg._messageId;
         _messageHandle = msg._messageHandle;
         _txnContext = msg._txnContext;
@@ -203,7 +228,14 @@ public class AMQMessage
 
     public ContentHeaderBody getContentHeaderBody() throws AMQException
     {
-        return _messageHandle.getContentHeaderBody(_messageId);
+        if (_contentHeaderBody != null)
+        {
+            return _contentHeaderBody;
+        }
+        else
+        {
+            return _messageHandle.getContentHeaderBody(_messageId);
+        }
     }
 
     public void setContentHeaderBody(ContentHeaderBody contentHeaderBody)
@@ -292,7 +324,12 @@ public class AMQMessage
                 {
                     _log.debug("Ref count on message " + _messageId + " is zero; removing message");
                 }
-                _messageHandle.removeMessage(_messageId);
+                // must check if the handle is null since there may be cases where we decide to throw away a message
+                // and the handle has not yet been constructed
+                if (_messageHandle != null)
+                {
+                    _messageHandle.removeMessage(_messageId);
+                }
             }
             catch (AMQException e)
             {
@@ -403,7 +440,7 @@ public class AMQMessage
     {
         return _messageHandle.isRedelivered();
     }
-    
+
     /**
      * Called when this message is delivered to a consumer. (used to
      * implement the 'immediate' flag functionality).
@@ -494,10 +531,10 @@ public class AMQMessage
         return buf;
     }
 
-    private ByteBuffer createEncodedReturnFrame(int channelId, int replyCode, String replyText)
+private ByteBuffer createEncodedReturnFrame(int channelId, int replyCode, String replyText) throws AMQException
     {
-        AMQFrame returnFrame = BasicReturnBody.createAMQFrame(channelId, replyCode, replyText, _publishBody.exchange,
-                                                              _publishBody.routingKey);
+        AMQFrame returnFrame = BasicReturnBody.createAMQFrame(channelId, replyCode, replyText, getPublishBody().exchange,
+                                                              getPublishBody().routingKey);
         ByteBuffer buf = ByteBuffer.allocate((int) returnFrame.getSize()); // XXX: Could cast be a problem?
         returnFrame.writePayload(buf);
         buf.flip();
diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
index df6520e0a8..06522740cc 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
@@ -23,6 +23,7 @@ import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
 import org.apache.qpid.server.queue.AMQMessage;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.RequiredDeliveryException;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.store.MessageStore;
 
 import java.util.List;
@@ -62,12 +63,6 @@ public class LocalTransactionalContext implements TransactionalContext
 
     public void deliver(AMQMessage message, AMQQueue queue) throws AMQException
     {
-        // don't create a transaction unless needed
-        if (message.isPersistent())
-        {
-            _txnBuffer.containsPersistentChanges();
-        }
-
         // A publication will result in the enlisting of several
         // TxnOps. The first is an op that will store the message.
         // Following that (and ordering is important), an op will
@@ -121,6 +116,11 @@ public class LocalTransactionalContext implements TransactionalContext
         // Not required in this transactional context
     }
 
+    public void messageProcessed(AMQProtocolSession protocolSession) throws AMQException
+    {
+        // Not required in this transactional context
+    }
+
     public void beginTranIfNecessary() throws AMQException
     {
         if (!_inTran)
@@ -134,16 +134,11 @@ public class LocalTransactionalContext implements TransactionalContext
     {
         if (_ackOp != null)
         {
-            _ackOp.consolidate();
-            if (_ackOp.checkPersistent())
-            {
-                _txnBuffer.containsPersistentChanges();
-            }
+            _ackOp.consolidate();            
             //already enlisted, after commit will reset regardless of outcome
             _ackOp = null;
         }
 
         _txnBuffer.commit();
-        //TODO: may need to return 'immediate' messages at this point
     }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java b/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
index 9c8e7c7002..8c3692a98d 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
@@ -21,6 +21,7 @@ import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.RequiredDeliveryException;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.ack.UnacknowledgedMessage;
 import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
 import org.apache.qpid.server.queue.AMQMessage;
@@ -60,7 +61,7 @@ public class NonTransactionalContext implements TransactionalContext
     {
         _channel = channel;
         _returnMessages = returnMessages;
-        _messageStore = messageStore;        
+        _messageStore = messageStore;
     }
 
     public void beginTranIfNecessary() throws AMQException
@@ -168,4 +169,9 @@ public class NonTransactionalContext implements TransactionalContext
             _inTran = false;
         }
     }
+
+    public void messageProcessed(AMQProtocolSession protocolSession) throws AMQException
+    {
+        _channel.processReturns(protocolSession);
+    }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java b/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
index 1ec216cc8b..74c7e51e69 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
@@ -12,6 +12,7 @@ import org.apache.qpid.AMQException;
 import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
 import org.apache.qpid.server.queue.AMQMessage;
 import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
 
 /**
  * @author Robert Greig (robert.j.greig@jpmorgan.com)
@@ -30,4 +31,6 @@ public interface TransactionalContext
                             UnacknowledgedMessageMap unacknowledgedMessageMap) throws AMQException;
 
     void messageFullyReceived(boolean persistent) throws AMQException;
+
+    void messageProcessed(AMQProtocolSession protocolSession) throws AMQException;
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java b/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java
index de37e134f0..63cb6f738b 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/txn/TxnBuffer.java
@@ -7,9 +7,9 @@
  * 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
@@ -22,7 +22,6 @@ package org.apache.qpid.server.txn;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.server.store.MessageStore;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -33,47 +32,20 @@ import java.util.List;
  */
 public class TxnBuffer
 {
-    private boolean _containsPersistentChanges = false;
-    private final MessageStore _store;
     private final List<TxnOp> _ops = new ArrayList<TxnOp>();
     private static final Logger _log = Logger.getLogger(TxnBuffer.class);
 
-    public TxnBuffer(MessageStore store)
-    {
-        _store = store;
-    }
-
-    public void containsPersistentChanges()
+    public TxnBuffer()
     {
-        _containsPersistentChanges = true;
     }
 
     public void commit() throws AMQException
     {
-        if (_containsPersistentChanges)
+        if (prepare())
         {
-            _log.debug("Begin Transaction.");
-            if (prepare())
+            for (TxnOp op : _ops)
             {
-                _log.debug("Transaction Succeeded");
-                for (TxnOp op : _ops)
-                {
-                    op.commit();
-                }
-            }
-            else
-            {
-                _log.debug("Transaction Failed");                
-            }
-        }
-        else
-        {
-            if (prepare())
-            {
-                for (TxnOp op : _ops)
-                {
-                    op.commit();
-                }
+                op.commit();
             }
         }
         _ops.clear();