QPID-3539: NoLocal should occur at the connection level and not the session level (0-10). Also fixed typing issue that prevented Java Broker understanding the no-local queue argument when encoded as String (as sent by Python tests).

Applied patch from Oleksandr Rudyy<orudyy@gmail.com> and myself.

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

2 files changed, 39 insertions, 12 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java b/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java
index c5d6bc203c..6598d07a84 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java
@@ -56,6 +56,7 @@ import org.apache.qpid.transport.MessageFlowMode;
 import org.apache.qpid.transport.MessageProperties;
 import org.apache.qpid.transport.MessageTransfer;
 import org.apache.qpid.transport.Method;
+import org.apache.qpid.transport.Session;
 import org.apache.qpid.transport.Struct;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
@@ -209,13 +210,13 @@ public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCr
             return false;
         }
 
-
-
-        if (_noLocal
-            && (entry.getMessage() instanceof MessageTransferMessage)
-            && ((MessageTransferMessage)entry.getMessage()).getSession() == _session)
+        if (_noLocal && entry.getMessage() instanceof MessageTransferMessage)
         {
-            return false;
+            Session messageSession= ((MessageTransferMessage)entry.getMessage()).getSession();
+            if (messageSession != null && messageSession.getConnection() == _session.getConnection())
+            {
+                return false;
+            }
         }
 
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
index c87919b478..754a233907 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
@@ -100,6 +100,11 @@ public class ServerSessionDelegate extends SessionDelegate
 {
     private static final Logger LOGGER = Logger.getLogger(ServerSessionDelegate.class);
 
+    /**
+     * No-local queue argument is used to support the no-local feature of Durable Subscribers.
+     */
+    private static final String QUEUE_ARGUMENT_NO_LOCAL = "no-local";
+
     public ServerSessionDelegate()
     {
 
@@ -963,13 +968,10 @@ public class ServerSessionDelegate extends SessionDelegate
 
                         if(method.hasArguments()  && method.getArguments() != null)
                         {
-                            if(method.getArguments().containsKey("no-local"))
+                            if(method.getArguments().containsKey(QUEUE_ARGUMENT_NO_LOCAL))
                             {
-                                Object no_local = method.getArguments().get("no-local");
-                                if(no_local instanceof Boolean && ((Boolean)no_local))
-                                {
-                                    queue.setNoLocal(true);
-                                }
+                                Object noLocal = method.getArguments().get(QUEUE_ARGUMENT_NO_LOCAL);
+                                queue.setNoLocal(convertBooleanValue(noLocal));
                             }
                         }
 
@@ -1079,6 +1081,30 @@ public class ServerSessionDelegate extends SessionDelegate
         }
     }
 
+    /**
+     * Converts a queue argument into a boolean value.  For compatibility with the C++
+     * and the clients, accepts with Boolean, String, or Number types.
+     * @param argValue  argument value.
+     *
+     * @return true if set
+     */
+    private boolean convertBooleanValue(Object argValue)
+    {
+        if(argValue instanceof Boolean && ((Boolean)argValue))
+        {
+            return true;
+        }
+        else if (argValue instanceof String && Boolean.parseBoolean((String)argValue))
+        {
+            return true;
+        }
+        else if (argValue instanceof Number && ((Number)argValue).intValue() != 0)
+        {
+            return true;
+        }
+        return false;
+    }
+
     protected AMQQueue createQueue(final String queueName,
                                    final QueueDeclare body,
                                    VirtualHost virtualHost,