QPID-2732

Only UNRELIABLE and AT_LEAST_ONCE is supported.
Currently the reliability mode is used only for determining the accept-mode.


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

3 files changed, 46 insertions, 2 deletions

diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
index b392bad879..6b4f6c1640 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
@@ -47,6 +47,8 @@ import org.apache.qpid.client.message.AMQMessageDelegateFactory;
 import org.apache.qpid.client.message.FieldTableSupport;
 import org.apache.qpid.client.message.MessageFactoryRegistry;
 import org.apache.qpid.client.message.UnprocessedMessage_0_10;
+import org.apache.qpid.client.messaging.address.Link;
+import org.apache.qpid.client.messaging.address.Link.Reliability;
 import org.apache.qpid.client.messaging.address.Node.ExchangeNode;
 import org.apache.qpid.client.messaging.address.Node.QueueNode;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
@@ -601,10 +603,16 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
                         (Map<? extends String, ? extends Object>) consumer.getDestination().getLink().getSubscription().getArgs());
             }
             
+            boolean acceptModeNone = false;
+            
+            if (consumer.getDestination().getLink() != null)
+            {
+                acceptModeNone = consumer.getDestination().getLink().getReliability() == Link.Reliability.UNRELIABLE;
+            }
             
             getQpidSession().messageSubscribe
                 (queueName.toString(), String.valueOf(tag),
-                 getAcknowledgeMode() == NO_ACKNOWLEDGE ? MessageAcceptMode.NONE : MessageAcceptMode.EXPLICIT,
+                 acceptModeNone ? MessageAcceptMode.NONE : MessageAcceptMode.EXPLICIT,
                  preAcquire ? MessageAcquireMode.PRE_ACQUIRED : MessageAcquireMode.NOT_ACQUIRED, null, 0, arguments,
                  consumer.isExclusive() ? Option.EXCLUSIVE : Option.NONE);
         }
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
index 00503cc650..e454a8eee4 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/AddressHelper.java
@@ -27,6 +27,7 @@ import java.util.Map;
 
 import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.client.AMQDestination.Binding;
+import org.apache.qpid.client.messaging.address.Link.Reliability;
 import org.apache.qpid.client.messaging.address.Link.Subscription;
 import org.apache.qpid.client.messaging.address.Node.ExchangeNode;
 import org.apache.qpid.client.messaging.address.Node.QueueNode;
@@ -262,7 +263,7 @@ public class AddressHelper
         }
     }
 
-    public Link getLink()
+    public Link getLink() throws Exception
     {
         Link link = new Link();
         link.setSubscription(new Subscription());
@@ -272,6 +273,25 @@ public class AddressHelper
                     : linkProps.getBoolean(DURABLE));
             link.setName(linkProps.getString(NAME));
 
+            String reliability = linkProps.getString(RELIABILITY);
+            if ( reliability != null)
+            {
+                if (reliability.equalsIgnoreCase("unreliable"))
+                {
+                    link.setReliability(Reliability.UNRELIABLE);
+                }
+                else if (reliability.equalsIgnoreCase("at-least-once"))
+                {
+                    link.setReliability(Reliability.AT_LEAST_ONCE);
+                }
+                else
+                {
+                    throw new Exception("The reliability mode '" + 
+                            reliability + "' is not yet supported");
+                }
+                
+            }
+            
             if (((Map) address.getOptions().get(LINK)).get(CAPACITY) instanceof Map)
             {
                 MapAccessor capacityProps = new MapAccessor(
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java
index a7d19d1bd5..f10c6f7fca 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/messaging/address/Link.java
@@ -20,6 +20,9 @@
  */
 package org.apache.qpid.client.messaging.address;
 
+import static org.apache.qpid.client.messaging.address.Link.Reliability.UNRELIABLE;
+import static org.apache.qpid.client.messaging.address.Link.Reliability.AT_LEAST_ONCE;
+
 import java.util.HashMap;
 import java.util.Map;
 
@@ -29,6 +32,8 @@ public class Link
 { 
     public enum FilterType { SQL92, XQUERY, SUBJECT }
     
+    public enum Reliability { UNRELIABLE, AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE }
+    
     protected String name;
     protected String _filter;
     protected FilterType _filterType = FilterType.SUBJECT;
@@ -38,7 +43,18 @@ public class Link
     protected int _producerCapacity = 0;
     protected Node node;
     protected Subscription subscription;
+    protected Reliability reliability = AT_LEAST_ONCE;
     
+    public Reliability getReliability()
+    {
+        return reliability;
+    }
+
+    public void setReliability(Reliability reliability)
+    {
+        this.reliability = reliability;
+    }
+
     public Node getNode()
     {
         return node;