QPID-1665: add a timer to ensure message acking is never delayed more than 1000 ms by default, this is configurable by qpid.session.max_ack_delay

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

4 files changed, 27 insertions, 1 deletions

diff --git a/qpid/java/common/Composite.tpl b/qpid/java/common/Composite.tpl
index 17cf846d8c..c46d0a12cc 100644
--- a/qpid/java/common/Composite.tpl
+++ b/qpid/java/common/Composite.tpl
@@ -145,6 +145,7 @@ if options or base == "Method":
   if base == "Method":
     out("""            case SYNC: this.setSync(true); break;
             case BATCH: this.setBatch(true); break;
+            case UNRELIABLE: this.setUnreliable(true); break;
 """)
   out("""            case NONE: break;
             default: throw new IllegalArgumentException("invalid option: " + _options[i]);
diff --git a/qpid/java/common/Option.tpl b/qpid/java/common/Option.tpl
index 776b211ad5..c22b35b999 100644
--- a/qpid/java/common/Option.tpl
+++ b/qpid/java/common/Option.tpl
@@ -37,5 +37,6 @@ for c in composites:
         options[option] = None
         out("    $option,\n")}
     BATCH,
+    UNRELIABLE,
     NONE
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Method.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Method.java
index 09cfd119be..611c742fb1 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Method.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Method.java
@@ -48,6 +48,7 @@ public abstract class Method extends Struct implements ProtocolEvent
     private boolean idSet = false;
     private boolean sync = false;
     private boolean batch = false;
+    private boolean unreliable = false;
 
     public final int getId()
     {
@@ -90,6 +91,16 @@ public abstract class Method extends Struct implements ProtocolEvent
         this.batch = value;
     }
 
+    public final boolean isUnreliable()
+    {
+        return unreliable;
+    }
+
+    final void setUnreliable(boolean value)
+    {
+        this.unreliable = value;
+    }
+
     public abstract boolean hasPayload();
 
     public Header getHeader()
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
index f94edcc655..4079097f96 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -384,7 +384,15 @@ public class Session extends SessionInvoker
         {
             copy = processed.copy();
         }
-        sessionCompleted(copy, options);
+
+        synchronized (commands)
+        {
+            if (state == DETACHED)
+            {
+                return;
+            }
+            sessionCompleted(copy, options);
+        }
     }
 
     void knownComplete(RangeSet kc)
@@ -484,6 +492,11 @@ public class Session extends SessionInvoker
 
             synchronized (commands)
             {
+                if (state == DETACHED && m.isUnreliable())
+                {
+                    return;
+                }
+
                 if (state != OPEN && state != CLOSED)
                 {
                     Waiter w = new Waiter(commands, timeout);