QPID-3713 : Implement producer side flow control for 0-10 in Java Broker

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

2 files changed, 25 insertions, 5 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/transport/Range.java b/java/common/src/main/java/org/apache/qpid/transport/Range.java
index f976337788..c47171dc4b 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Range.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Range.java
@@ -185,6 +185,12 @@ public abstract class Range implements RangeSet
             }
         }
 
+        public String toString()
+        {
+            return "[" + point + ", " + point + "]";
+        }
+
+
     }
 
     private static class RangeImpl extends Range
@@ -283,7 +289,7 @@ public abstract class Range implements RangeSet
             return range;
         }
 
-        @Override
+
         public void remove()
         {
             throw new UnsupportedOperationException();
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 3e823ba6fe..d391181217 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
@@ -61,7 +61,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 public class Session extends SessionInvoker
 {
     private static final Logger log = Logger.get(Session.class);
-
+    
     public enum State { NEW, DETACHED, RESUMING, OPEN, CLOSING, CLOSED }
 
     static class DefaultSessionListener implements SessionListener
@@ -96,6 +96,9 @@ public class Session extends SessionInvoker
     private final long timeout = Long.getLong(ClientProperties.QPID_SYNC_OP_TIMEOUT,
                                         Long.getLong(ClientProperties.AMQJ_DEFAULT_SYNCWRITE_TIMEOUT,
                                                      ClientProperties.DEFAULT_SYNC_OPERATION_TIMEOUT));
+    private final long blockedSendTimeout = Long.getLong("qpid.flow_control_wait_failure", timeout);
+    private long blockedSendReportingPeriod = Long.getLong("qpid.flow_control_wait_notify_period",5000L);
+
     private boolean autoSync = false;
 
     private boolean incomingInit;
@@ -228,10 +231,21 @@ public class Session extends SessionInvoker
         {
             try
             {
-                if (!credit.tryAcquire(timeout, TimeUnit.MILLISECONDS))
+                long wait = blockedSendTimeout > blockedSendReportingPeriod ? blockedSendReportingPeriod :
+                           blockedSendTimeout;
+                long totalWait = 1L;
+                while(totalWait <= blockedSendTimeout && !credit.tryAcquire(wait, TimeUnit.MILLISECONDS))
+                {
+                    totalWait+=wait;
+                    log.warn("Message send delayed by " + (totalWait)/1000 + "s due to broker enforced flow control");
+
+
+                }
+                if(totalWait > blockedSendTimeout)
                 {
+                    log.error("Message send failed due to timeout waiting on broker enforced flow control");
                     throw new SessionException
-                        ("timed out waiting for message credit");
+                            ("timed out waiting for message credit");
                 }
             }
             catch (InterruptedException e)
@@ -815,7 +829,7 @@ public class Session extends SessionInvoker
             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, commands);
+                log.debug("%s   waiting for[%d]: %d, %s", this, point, maxComplete, Arrays.asList(commands));
                 w.await();
             }