QPID-933 : performance tweaks

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

2 files changed, 54 insertions, 11 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index d17572ad77..eb776ba786 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -182,6 +182,46 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                 _fastAccessConsumers[i] = null;
             }
         }
+
+
+        public void acknowledgeDelivered()
+        {
+
+            for(int i = 0; i<16; i++)
+            {
+                final BasicMessageConsumer c = _fastAccessConsumers[i];
+                if(c != null)
+                {
+                    c.acknowledgeDelivered();
+                }
+            }
+            if(!_slowAccessConsumers.isEmpty())
+            {
+                for (Iterator<BasicMessageConsumer> i = _slowAccessConsumers.values().iterator(); i.hasNext();)
+                {
+                    i.next().acknowledgeDelivered();
+                }
+            }
+        }
+
+        public void acknowledge() throws JMSException
+        {
+            for(int i = 0; i<16; i++)
+            {
+                final BasicMessageConsumer c = _fastAccessConsumers[i];
+                if(c != null)
+                {
+                    c.acknowledge();
+                }
+            }
+            if(!_slowAccessConsumers.isEmpty())
+            {
+                for (Iterator<BasicMessageConsumer> i = _slowAccessConsumers.values().iterator(); i.hasNext();)
+                {
+                    i.next().acknowledge();
+                }
+            }
+        }
     }
 
 
@@ -500,10 +540,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
             throw new IllegalStateException("Session is already closed");
         }
 
-        for (BasicMessageConsumer consumer : _consumers.values())
-        {
-            consumer.acknowledge();
-        }
+        _consumers.acknowledge();
     }
 
     /**
@@ -725,12 +762,9 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                     // We only need to find the highest value and ack that as commit is session level.
                     Long lastTag = -1L;
 
-                    for (Iterator<BasicMessageConsumer> i = _consumers.values().iterator(); i.hasNext();)
-                    {
-                        i.next().acknowledgeDelivered();
-                    }
+                    _consumers.acknowledgeDelivered();
 
-                    if (_transacted)
+                    if (_transacted && !_removedConsumers.isEmpty())
                     {
                         // Do the above, but for consumers which have been de-registered since the
                         // last commit
diff --git a/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java b/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
index 87df7e1337..4de6a1d410 100644
--- a/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
+++ b/java/client/src/main/java/org/apache/qpid/client/message/AbstractJMSMessageFactory.java
@@ -71,8 +71,17 @@ public abstract class AbstractJMSMessageFactory implements MessageFactory
             while (it.hasNext())
             {
                 ContentBody cb = (ContentBody) it.next();
-                data.put(cb.payload);
-                cb.payload.release();
+                final ByteBuffer payload = cb.payload;
+                if(payload.isDirect() || payload.isReadOnly())
+                {
+                    data.put(payload);
+                }
+                else
+                {
+                    data.put(payload.array(), payload.arrayOffset(), payload.limit());
+                }
+
+                payload.release();
             }
 
             data.flip();