QPID-3720 : [Java Broker] Implement Message Grouping

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

2 files changed, 17 insertions, 2 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java b/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
index 84e4056f4d..f64164c10b 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
@@ -166,6 +166,11 @@ public abstract class AMQTypedValue
 
     private static final class IntTypedValue extends AMQTypedValue
     {
+        @Override
+        public String toString()
+        {
+            return "[INT: " + String.valueOf(_value) + "]";
+        }
 
         private final int _value;
 
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Session.java b/java/common/src/main/java/org/apache/qpid/transport/Session.java
index d391181217..5a9ea73cae 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -44,6 +44,7 @@ import static org.apache.qpid.util.Serial.max;
 import static org.apache.qpid.util.Strings.toUTF8;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -828,8 +829,17 @@ public class Session extends SessionInvoker
             Waiter w = new Waiter(commands, timeout);
             while (w.hasTime() && state != CLOSED && lt(maxComplete, point))
             {
-                checkFailoverRequired("Session sync was interrupted by failover.");
-                log.debug("%s   waiting for[%d]: %d, %s", this, point, maxComplete, Arrays.asList(commands));
+                checkFailoverRequired("Session sync was interrupted by failover.");                               
+                if(log.isDebugEnabled())
+                {
+                    List<Method> waitingFor =
+                            Arrays.asList(commands)
+                                  .subList(mod(maxComplete,commands.length),
+                                           mod(commandsOut-1, commands.length) < mod(maxComplete, commands.length)
+                                             ? commands.length-1
+                                             : mod(commandsOut-1, commands.length));
+                    log.debug("%s   waiting for[%d]: %d, %s", this, point, maxComplete, waitingFor);
+                }
                 w.await();
             }