QPID-790 : Performance Improvements

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

6 files changed, 86 insertions, 25 deletions

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;