QPID-790 : Performance Improvements

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

19 files changed, 192 insertions, 63 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 1bf0cd027a..10184a79e5 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
@@ -35,6 +35,7 @@ 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.exchange.Exchange;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.queue.*;
 import org.apache.qpid.server.store.MessageStore;
@@ -199,11 +200,12 @@ public class AMQChannel
         _prefetch_HighWaterMark = prefetchCount;
     }
 
-    public void setPublishFrame(MessagePublishInfo info, AMQProtocolSession publisher) throws AMQException
+    public void setPublishFrame(MessagePublishInfo info, AMQProtocolSession publisher, final Exchange e) throws AMQException
     {
 
         _currentMessage = new AMQMessage(_messageStore.getNewMessageId(), info, _txnContext);
         _currentMessage.setPublisher(publisher);
+        _currentMessage.setExchange(e);
     }
 
     public void publishContentHeader(ContentHeaderBody contentHeaderBody, AMQProtocolSession protocolSession)
@@ -285,7 +287,7 @@ public class AMQChannel
     {
         try
         {
-            _exchanges.routeContent(_currentMessage);
+            _currentMessage.route();            
         }
         catch (NoRouteException e)
         {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
index 75be86a387..19172b98f3 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
@@ -239,7 +239,7 @@ public class DestWildExchange extends AbstractExchange
     {
         MessagePublishInfo info = payload.getMessagePublishInfo();
 
-        final AMQShortString routingKey = normalize(info.getRoutingKey());
+        final AMQShortString routingKey = info.getRoutingKey();
 
         List<AMQQueue> queues = getMatchedQueues(routingKey);
         // if we have no registered queues we have nothing to do
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
index 66afc61751..687ec33ba0 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
@@ -91,7 +91,8 @@ public class BasicPublishMethodHandler  implements StateAwareMethodListener<Basi
             }
 
             MessagePublishInfo info = session.getMethodRegistry().getProtocolVersionMethodConverter().convertToInfo(body);
-            channel.setPublishFrame(info, session);
+            info.setExchange(exchange);
+            channel.setPublishFrame(info, session, e);
         }
     }
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java
index 1bfe1e3d35..98c77d8d32 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java
@@ -72,7 +72,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
     public void writeDeliver(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)

             throws AMQException

     {

-        ByteBuffer deliver = createEncodedDeliverFrame(message, channelId, deliveryTag, consumerTag);

+        AMQDataBlock deliver = createEncodedDeliverFrame(message, channelId, deliveryTag, consumerTag);

         AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

                                                                       message.getContentHeaderBody());

 

@@ -127,7 +127,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
         final StoreContext storeContext = message.getStoreContext();

         final long messageId = message.getMessageId();

 

-        ByteBuffer deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);

+        AMQDataBlock deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);

 

 

         AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

@@ -171,7 +171,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
     }

 

 

-    private ByteBuffer createEncodedDeliverFrame(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)

+    private AMQDataBlock createEncodedDeliverFrame(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)

             throws AMQException

     {

         final MessagePublishInfo pb = message.getMessagePublishInfo();

@@ -187,10 +187,10 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
         AMQFrame deliverFrame = deliverBody.generateFrame(channelId);

 

 

-        return deliverFrame.toByteBuffer();

+        return deliverFrame;

     }

 

-    private ByteBuffer createEncodedGetOkFrame(AMQMessage message, int channelId, long deliveryTag, int queueSize)

+    private AMQDataBlock createEncodedGetOkFrame(AMQMessage message, int channelId, long deliveryTag, int queueSize)

             throws AMQException

     {

         final MessagePublishInfo pb = message.getMessagePublishInfo();

@@ -205,7 +205,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
                                                     queueSize);

         AMQFrame getOkFrame = getOkBody.generateFrame(channelId);

 

-        return getOkFrame.toByteBuffer();

+        return getOkFrame;

     }

 

     public byte getProtocolMinorVersion()

@@ -218,7 +218,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
         return getProtocolSession().getProtocolMajorVersion();

     }

 

-    private ByteBuffer createEncodedReturnFrame(AMQMessage message, int channelId, int replyCode, AMQShortString replyText) throws AMQException

+    private AMQDataBlock createEncodedReturnFrame(AMQMessage message, int channelId, int replyCode, AMQShortString replyText) throws AMQException

     {

         MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0);

         BasicReturnBody basicReturnBody =

@@ -228,13 +228,13 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
                                                      message.getMessagePublishInfo().getRoutingKey());

         AMQFrame returnFrame = basicReturnBody.generateFrame(channelId);

 

-        return returnFrame.toByteBuffer();

+        return returnFrame;

     }

 

     public void writeReturn(AMQMessage message, int channelId, int replyCode, AMQShortString replyText)

             throws AMQException

     {

-        ByteBuffer returnFrame = createEncodedReturnFrame(message, channelId, replyCode, replyText);

+        AMQDataBlock returnFrame = createEncodedReturnFrame(message, channelId, replyCode, replyText);

 

         AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

                                                                       message.getContentHeaderBody());

diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java
index 0bc2fcf6f7..b14f03e617 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java
@@ -16,6 +16,7 @@ import org.apache.qpid.AMQException;
 

 public class ProtocolOutputConverterImpl implements ProtocolOutputConverter

 {

+    private static final MethodRegistry METHOD_REGISTRY = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);

 

 

     public static Factory getInstanceFactory()

@@ -46,9 +47,10 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
     public void writeDeliver(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)

             throws AMQException

     {

-        ByteBuffer deliver = createEncodedDeliverFrame(message, channelId, deliveryTag, consumerTag);

+        AMQDataBlock deliver = createEncodedDeliverFrame(message, channelId, deliveryTag, consumerTag);

+        final ContentHeaderBody contentHeaderBody = message.getContentHeaderBody();

         AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

-                                                                      message.getContentHeaderBody());

+                                                                      contentHeaderBody);

 

         final AMQMessageHandle messageHandle = message.getMessageHandle();

         final StoreContext storeContext = message.getStoreContext();

@@ -101,7 +103,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
         final StoreContext storeContext = message.getStoreContext();

         final long messageId = message.getMessageId();

 

-        ByteBuffer deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);

+        AMQFrame deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);

 

 

         AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

@@ -145,41 +147,54 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
     }

 

 

-    private ByteBuffer createEncodedDeliverFrame(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)

+    private AMQDataBlock createEncodedDeliverFrame(AMQMessage message, final int channelId, final long deliveryTag, final AMQShortString consumerTag)

             throws AMQException

     {

         final MessagePublishInfo pb = message.getMessagePublishInfo();

         final AMQMessageHandle messageHandle = message.getMessageHandle();

 

-        MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);

-        BasicDeliverBody deliverBody =

-                methodRegistry.createBasicDeliverBody(consumerTag,

-                                                      deliveryTag,

-                                                      messageHandle.isRedelivered(),

-                                                      pb.getExchange(),

-                                                      pb.getRoutingKey());

-        AMQFrame deliverFrame = deliverBody.generateFrame(channelId);

 

+        final boolean isRedelivered = messageHandle.isRedelivered();

+        final AMQShortString exchangeName = pb.getExchange();

+        final AMQShortString routingKey = pb.getRoutingKey();

+

+        final AMQDataBlock returnBlock = new DeferredDataBlock()

+        {

+

+            protected AMQDataBlock createAMQDataBlock()

+            {

+                BasicDeliverBody deliverBody =

+                        METHOD_REGISTRY.createBasicDeliverBody(consumerTag,

+                                                              deliveryTag,

+                                                              isRedelivered,

+                                                              exchangeName,

+                                                              routingKey);

+                AMQFrame deliverFrame = deliverBody.generateFrame(channelId);

+

+

+                return deliverFrame;

 

-        return deliverFrame.toByteBuffer();

+            }

+        };

+        return returnBlock;

     }

 

-    private ByteBuffer createEncodedGetOkFrame(AMQMessage message, int channelId, long deliveryTag, int queueSize)

+    private AMQFrame createEncodedGetOkFrame(AMQMessage message, int channelId, long deliveryTag, int queueSize)

             throws AMQException

     {

         final MessagePublishInfo pb = message.getMessagePublishInfo();

         final AMQMessageHandle messageHandle = message.getMessageHandle();

 

-        MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);

+

         BasicGetOkBody getOkBody =

-                methodRegistry.createBasicGetOkBody(deliveryTag,

+                METHOD_REGISTRY.createBasicGetOkBody(deliveryTag,

                                                     messageHandle.isRedelivered(),

                                                     pb.getExchange(),

                                                     pb.getRoutingKey(),

                                                     queueSize);

         AMQFrame getOkFrame = getOkBody.generateFrame(channelId);

 

-        return getOkFrame.toByteBuffer();

+        return getOkFrame;

     }

 

     public byte getProtocolMinorVersion()

@@ -192,23 +207,23 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
         return getProtocolSession().getProtocolMajorVersion();

     }

 

-    private ByteBuffer createEncodedReturnFrame(AMQMessage message, int channelId, int replyCode, AMQShortString replyText) throws AMQException

+    private AMQDataBlock createEncodedReturnFrame(AMQMessage message, int channelId, int replyCode, AMQShortString replyText) throws AMQException

     {

-        MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);

+

         BasicReturnBody basicReturnBody =

-                methodRegistry.createBasicReturnBody(replyCode,

+                METHOD_REGISTRY.createBasicReturnBody(replyCode,

                                                      replyText,

                                                      message.getMessagePublishInfo().getExchange(),

                                                      message.getMessagePublishInfo().getRoutingKey());

         AMQFrame returnFrame = basicReturnBody.generateFrame(channelId);

 

-        return returnFrame.toByteBuffer();

+        return returnFrame;

     }

 

     public void writeReturn(AMQMessage message, int channelId, int replyCode, AMQShortString replyText)

             throws AMQException

     {

-        ByteBuffer returnFrame = createEncodedReturnFrame(message, channelId, replyCode, replyText);

+        AMQDataBlock returnFrame = createEncodedReturnFrame(message, channelId, replyCode, replyText);

 

         AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

                                                                       message.getContentHeaderBody());

@@ -252,8 +267,8 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
 

     public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag)

     {

-        MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);

-        BasicCancelOkBody basicCancelOkBody = methodRegistry.createBasicCancelOkBody(consumerTag);

+

+        BasicCancelOkBody basicCancelOkBody = METHOD_REGISTRY.createBasicCancelOkBody(consumerTag);

         writeFrame(basicCancelOkBody.generateFrame(channelId));

 

     }

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 d9a9d2273b..80158779b2 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
@@ -35,6 +35,7 @@ import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.StoreContext;
 import org.apache.qpid.server.txn.TransactionalContext;
+import org.apache.qpid.server.exchange.Exchange;
 
 import java.util.HashMap;
 import java.util.HashSet;
@@ -84,6 +85,10 @@ public class AMQMessage
 
     private final int hashcode = System.identityHashCode(this);
 
+    private Exchange _exchange;
+    private static final boolean SYNCED_CLOCKS =
+            ApplicationRegistry.getInstance().getConfiguration().getBoolean("advanced.synced-clocks", false);
+
 
     public String debugIdentity()
     {
@@ -97,7 +102,7 @@ public class AMQMessage
         long timestamp =
                 ((BasicContentHeaderProperties) _transientMessageData.getContentHeaderBody().properties).getTimestamp();
 
-        if (ApplicationRegistry.getInstance().getConfiguration().getBoolean("advanced.synced-clocks", false))
+        if (SYNCED_CLOCKS)
         {
             _expiration = expiration;
         }
@@ -126,6 +131,16 @@ public class AMQMessage
         return _referenceCount.get() > 0;
     }
 
+    public void setExchange(final Exchange exchange)
+    {
+        _exchange = exchange;
+    }
+
+    public void route() throws AMQException
+    {
+        _exchange.route(this);
+    }
+
     /**
      * Used to iterate through all the body frames associated with this message. Will not keep all the data in memory
      * therefore is memory-efficient.
diff --git a/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java b/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java
index 8e5879a51e..7e2d56b460 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java
@@ -589,6 +589,11 @@ public class DestWildExchangeTest extends TestCase
             return null;
         }
 
+        public void setExchange(AMQShortString exchange)
+        {
+                        
+        }
+
         public boolean isImmediate()
         {
             return false;
diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
index 81b0ae2213..fbd9e65480 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
@@ -242,6 +242,11 @@ public class AMQQueueAlertTest extends TestCase
                 return null;
             }
 
+            public void setExchange(AMQShortString exchange)
+            {
+                //To change body of implemented methods use File | Settings | File Templates.
+            }
+
             public boolean isImmediate()
             {
                 return immediate;
diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
index d86c90bdae..e72e1bf1f0 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
@@ -234,6 +234,11 @@ public class AMQQueueMBeanTest extends TestCase
                 return null;
             }
 
+            public void setExchange(AMQShortString exchange)
+            {
+                //To change body of implemented methods use File | Settings | File Templates.
+            }
+
             public boolean isImmediate()
             {
                 return immediate;
diff --git a/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java
index 5ec62ede93..7b6699b783 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java
@@ -24,7 +24,7 @@ import org.apache.mina.common.ByteBuffer;
 
 public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock
 {
-    private ByteBuffer _encodedBlock;
+    private AMQDataBlock _firstFrame;
 
     private AMQDataBlock[] _blocks;
 
@@ -39,10 +39,10 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD
      * @param encodedBlock already-encoded data
      * @param blocks some blocks to be encoded.
      */
-    public CompositeAMQDataBlock(ByteBuffer encodedBlock, AMQDataBlock[] blocks)
+    public CompositeAMQDataBlock(AMQDataBlock encodedBlock, AMQDataBlock[] blocks)
     {
         this(blocks);
-        _encodedBlock = encodedBlock;
+        _firstFrame = encodedBlock;
     }
 
     public AMQDataBlock[] getBlocks()
@@ -50,9 +50,9 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD
         return _blocks;
     }
 
-    public ByteBuffer getEncodedBlock()
+    public AMQDataBlock getFirstFrame()
     {
-        return _encodedBlock;
+        return _firstFrame;
     }
 
     public long getSize()
@@ -62,19 +62,18 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD
         {
             frameSize += _blocks[i].getSize();
         }
-        if (_encodedBlock != null)
+        if (_firstFrame != null)
         {
-            _encodedBlock.rewind();
-            frameSize += _encodedBlock.remaining();
+            frameSize += _firstFrame.getSize();
         }
         return frameSize;
     }
 
     public void writePayload(ByteBuffer buffer)
     {
-        if (_encodedBlock != null)
+        if (_firstFrame != null)
         {
-            buffer.put(_encodedBlock);
+            _firstFrame.writePayload(buffer);
         }
         for (int i = 0; i < _blocks.length; i++)
         {
@@ -91,7 +90,7 @@ public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQD
         else
         {
             StringBuilder buf = new StringBuilder(this.getClass().getName());
-            buf.append("{encodedBlock=").append(_encodedBlock);
+            buf.append("{encodedBlock=").append(_firstFrame);
             for (int i = 0 ; i < _blocks.length; i++)
             {
                 buf.append(" ").append(i).append("=[").append(_blocks[i].toString()).append("]");
diff --git a/java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java
new file mode 100644
index 0000000000..f6795ff200
--- /dev/null
+++ b/java/common/src/main/java/org/apache/qpid/framing/DeferredDataBlock.java
@@ -0,0 +1,50 @@
+package org.apache.qpid.framing;
+
+import org.apache.mina.common.ByteBuffer;
+
+/*
+*
+* 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.
+*
+*/
+public abstract class DeferredDataBlock extends AMQDataBlock
+{
+    private AMQDataBlock _underlyingDataBlock;
+
+
+    public long getSize()
+    {
+        if(_underlyingDataBlock == null)
+        {
+            _underlyingDataBlock = createAMQDataBlock();
+        }
+        return _underlyingDataBlock.getSize();
+    }
+
+    public void writePayload(ByteBuffer buffer)
+    {
+        if(_underlyingDataBlock == null)
+        {
+            _underlyingDataBlock = createAMQDataBlock();
+        }
+        _underlyingDataBlock.writePayload(buffer);
+    }
+
+    abstract protected AMQDataBlock createAMQDataBlock();
+
+}
diff --git a/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java b/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java
index 26c048e34a..f8cf3f3011 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java
@@ -25,7 +25,7 @@ import org.apache.mina.common.ByteBuffer;
 

 public class SmallCompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock

 {

-    private ByteBuffer _encodedBlock;

+    private AMQDataBlock _firstFrame;

 

     private AMQDataBlock _block;

 

@@ -40,10 +40,10 @@ public class SmallCompositeAMQDataBlock extends AMQDataBlock implements Encodabl
      * @param encodedBlock already-encoded data

      * @param block a block to be encoded.

      */

-    public SmallCompositeAMQDataBlock(ByteBuffer encodedBlock, AMQDataBlock block)

+    public SmallCompositeAMQDataBlock(AMQDataBlock encodedBlock, AMQDataBlock block)

     {

         this(block);

-        _encodedBlock = encodedBlock;

+        _firstFrame = encodedBlock;

     }

 

     public AMQDataBlock getBlock()

@@ -51,28 +51,28 @@ public class SmallCompositeAMQDataBlock extends AMQDataBlock implements Encodabl
         return _block;

     }

 

-    public ByteBuffer getEncodedBlock()

+    public AMQDataBlock getFirstFrame()

     {

-        return _encodedBlock;

+        return _firstFrame;

     }

 

     public long getSize()

     {

         long frameSize = _block.getSize();

 

-        if (_encodedBlock != null)

+        if (_firstFrame != null)

         {

-            _encodedBlock.rewind();

-            frameSize += _encodedBlock.remaining();

+

+            frameSize += _firstFrame.getSize();

         }

         return frameSize;

     }

 

     public void writePayload(ByteBuffer buffer)

     {

-        if (_encodedBlock != null)

+        if (_firstFrame != null)

         {

-            buffer.put(_encodedBlock);

+            _firstFrame.writePayload(buffer);

         }

         _block.writePayload(buffer);

 

@@ -87,7 +87,7 @@ public class SmallCompositeAMQDataBlock extends AMQDataBlock implements Encodabl
         else

         {

             StringBuilder buf = new StringBuilder(this.getClass().getName());

-            buf.append("{encodedBlock=").append(_encodedBlock);

+            buf.append("{encodedBlock=").append(_firstFrame);

 

             buf.append(" _block=[").append(_block.toString()).append("]");

 

diff --git a/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java b/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java
index 706499c1b0..49c28bb06b 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/abstraction/MessagePublishInfo.java
@@ -27,6 +27,8 @@ public interface MessagePublishInfo
 

     public AMQShortString getExchange();

 

+    public void setExchange(AMQShortString exchange);

+

     public boolean isImmediate();

 

     public boolean isMandatory();

diff --git a/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java b/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java
index de0007c132..d8b6b25b92 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java
@@ -67,7 +67,7 @@ public class MethodConverter_0_9 extends AbstractMethodConverter implements Prot
         final AMQShortString exchange = publishBody.getExchange();

         final AMQShortString routingKey = publishBody.getRoutingKey();

 

-        return new MethodConverter_0_9.MessagePublishInfoImpl(exchange == null ? null : exchange.intern(),

+        return new MethodConverter_0_9.MessagePublishInfoImpl(exchange,

                                           publishBody.getImmediate(),

                                           publishBody.getMandatory(),

                                           routingKey == null ? null : routingKey.intern());

@@ -87,7 +87,7 @@ public class MethodConverter_0_9 extends AbstractMethodConverter implements Prot
 

     private static class MessagePublishInfoImpl implements MessagePublishInfo

     {

-        private final AMQShortString _exchange;

+        private AMQShortString _exchange;

         private final boolean _immediate;

         private final boolean _mandatory;

         private final AMQShortString _routingKey;

@@ -108,6 +108,11 @@ public class MethodConverter_0_9 extends AbstractMethodConverter implements Prot
             return _exchange;

         }

 

+        public void setExchange(AMQShortString exchange)

+        {

+            _exchange = exchange;

+        }

+

         public boolean isImmediate()

         {

             return _immediate;

diff --git a/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java b/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
index 7a13af8a43..b1be49a350 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
@@ -107,7 +107,7 @@ public class MethodConverter_8_0 extends AbstractMethodConverter implements Prot
 

     private static class MessagePublishInfoImpl implements MessagePublishInfo

     {

-        private final AMQShortString _exchange;

+        private AMQShortString _exchange;

         private final boolean _immediate;

         private final boolean _mandatory;

         private final AMQShortString _routingKey;

@@ -128,6 +128,11 @@ public class MethodConverter_8_0 extends AbstractMethodConverter implements Prot
             return _exchange;

         }

 

+        public void setExchange(AMQShortString exchange)

+        {

+            _exchange = exchange;

+        }

+

         public boolean isImmediate()

         {

             return _immediate;

diff --git a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
index 10189a8017..42d9cccb4f 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
@@ -117,6 +117,11 @@ public class TxAckTest extends TestCase
                         return null;
                     }
 
+                    public void setExchange(AMQShortString exchange)
+                    {
+                        //To change body of implemented methods use File | Settings | File Templates.
+                    }
+
                     public boolean isImmediate()
                     {
                         return false;
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
index 790607e268..96be579d2a 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
@@ -107,6 +107,11 @@ public class AckTest extends TestCase
                     return new AMQShortString("someExchange");
                 }
 
+                public void setExchange(AMQShortString exchange)
+                {
+                    //To change body of implemented methods use File | Settings | File Templates.
+                }
+
                 public boolean isImmediate()
                 {
                     return false;
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
index 812aec6a5d..521bedeccd 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
@@ -70,6 +70,11 @@ class MessageTestHelper extends TestCase
                 return null;
             }
 
+            public void setExchange(AMQShortString exchange)
+            {
+                //To change body of implemented methods use File | Settings | File Templates.
+            }
+
             public boolean isImmediate()
             {
                 return immediate;
diff --git a/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java b/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
index ab6d9742e4..28cc89353b 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
@@ -61,6 +61,11 @@ public class TestReferenceCounting extends TestCase
                 return null;
             }
 
+            public void setExchange(AMQShortString exchange)
+            {
+                //To change body of implemented methods use File | Settings | File Templates.
+            }
+
             public boolean isImmediate()
             {
                 return false;