diff options
Diffstat (limited to 'qpid/java/broker/src')
2 files changed, 39 insertions, 12 deletions
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java index c5d6bc203c..6598d07a84 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java +++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java index c87919b478..754a233907 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java +++ b/qpid/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, |