You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2014/01/07 18:58:42 UTC
svn commit: r1556292 - in /qpid/trunk/qpid/java:
broker-core/src/main/java/org/apache/qpid/server/virtualhost/
broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/
client/src/main/java/org/apache/qpid/client/ cli...
Author: kwall
Date: Tue Jan 7 17:58:41 2014
New Revision: 1556292
URL: http://svn.apache.org/r1556292
Log:
QPID-5420: Restore ability to consume using BURLs specifying default exchange.
* Java Broker: Changed AbstractVirtualHost so that createExchange throws ExchangeExistException before checking for a reserved exchnage name. The effect will be that the Java Broker will again accept active declaration of the built in exchanges (amq.*, qpid.* and default).
* Java Broker: Changed the 0-8..0-9-1 ExchangeBoundHandler so that a null exchnage name is treated to mean the default exchange. This matches the behaviour of ServerSessionDelegate#exchangeBound() on the 0-10 path. This allows the Java client to query bindings on the default exchange.
* Client: Changed AbstractAMQMessageDelegate.java so that 0-10 knows the type of the default exchange when populating the JMSDestination on received messages.
* Client: Introduced system property qpid.bind_queues system property so that the exchange/queue bind side effect can be suppressed on consumer creation. Like qid.declare_exchanges and declare_queues, this system propery has effect when using BURLs. Might be useful if using a new client with older broker.
* Added new system tests.
Modified:
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeBoundHandler.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/message/JMSDestinationTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/DynamicQueueExchangeCreateTest.java
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java Tue Jan 7 17:58:41 2014
@@ -563,11 +563,6 @@ public abstract class AbstractVirtualHos
String alternateExchangeName)
throws AMQException
{
-
- if(_exchangeRegistry.isReservedExchangeName(name))
- {
- throw new ReservedExchangeNameException(name);
- }
synchronized (_exchangeRegistry)
{
Exchange existing;
@@ -575,6 +570,11 @@ public abstract class AbstractVirtualHos
{
throw new ExchangeExistsException(name,existing);
}
+ if(_exchangeRegistry.isReservedExchangeName(name))
+ {
+ throw new ReservedExchangeNameException(name);
+ }
+
Exchange alternateExchange;
if(alternateExchangeName != null)
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeBoundHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeBoundHandler.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeBoundHandler.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/ExchangeBoundHandler.java Tue Jan 7 17:58:41 2014
@@ -29,7 +29,6 @@ import org.apache.qpid.server.protocol.v
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.protocol.v0_8.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.protocol.v0_8.state.AMQStateManager;
import org.apache.qpid.server.protocol.v0_8.state.StateAwareMethodListener;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -80,13 +79,9 @@ public class ExchangeBoundHandler implem
channel.sync();
- AMQShortString exchangeName = body.getExchange();
+ AMQShortString exchangeName = body.getExchange() == null ? AMQShortString.EMPTY_STRING : body.getExchange();
AMQShortString queueName = body.getQueue();
AMQShortString routingKey = body.getRoutingKey();
- if (exchangeName == null)
- {
- throw new AMQException("Exchange exchange must not be null");
- }
Exchange exchange = virtualHost.getExchange(exchangeName.toString());
ExchangeBoundOkBody response;
if (exchange == null)
@@ -94,7 +89,7 @@ public class ExchangeBoundHandler implem
response = methodRegistry.createExchangeBoundOkBody(EXCHANGE_NOT_FOUND,
- new AMQShortString("Exchange " + exchangeName + " not found"));
+ new AMQShortString("Exchange '" + exchangeName + "' not found"));
}
else if (routingKey == null)
{
@@ -119,7 +114,7 @@ public class ExchangeBoundHandler implem
{
response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode
- new AMQShortString("Queue " + queueName + " not found")); // replyText
+ new AMQShortString("Queue '" + queueName + "' not found")); // replyText
}
else
{
@@ -133,7 +128,7 @@ public class ExchangeBoundHandler implem
{
response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_BOUND, // replyCode
- new AMQShortString("Queue " + queueName + " not bound to exchange " + exchangeName)); // replyText
+ new AMQShortString("Queue '" + queueName + "' not bound to exchange '" + exchangeName + "'")); // replyText
}
}
}
@@ -145,7 +140,7 @@ public class ExchangeBoundHandler implem
{
response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode
- new AMQShortString("Queue " + queueName + " not found")); // replyText
+ new AMQShortString("Queue '" + queueName + "' not found")); // replyText
}
else
{
@@ -159,8 +154,8 @@ public class ExchangeBoundHandler implem
else
{
- String message = "Queue " + queueName + " not bound with routing key " +
- body.getRoutingKey() + " to exchange " + exchangeName;
+ String message = "Queue '" + queueName + "' not bound with routing key '" +
+ body.getRoutingKey() + "' to exchange '" + exchangeName + "'";
if(message.length()>255)
{
@@ -183,8 +178,8 @@ public class ExchangeBoundHandler implem
{
response = methodRegistry.createExchangeBoundOkBody(NO_QUEUE_BOUND_WITH_RK, // replyCode
- new AMQShortString("No queue bound with routing key " + body.getRoutingKey() +
- " to exchange " + exchangeName)); // replyText
+ new AMQShortString("No queue bound with routing key '" + body.getRoutingKey() +
+ "' to exchange '" + exchangeName + "'")); // replyText
}
}
session.writeFrame(response.generateFrame(channelId));
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Tue Jan 7 17:58:41 2014
@@ -109,7 +109,7 @@ public abstract class AMQSession<C exten
/** System property to enable failure if strict AMQP compliance is violated. */
public static final String STRICT_AMQP_FATAL = "STRICT_AMQP_FATAL";
- /** Strickt AMQP failure default. */
+ /** Strict AMQP failure default. */
public static final String STRICT_AMQP_FATAL_DEFAULT = "true";
/** System property to enable immediate message prefetching. */
@@ -124,6 +124,9 @@ public abstract class AMQSession<C exten
private final boolean _declareExchanges =
Boolean.parseBoolean(System.getProperty(ClientProperties.QPID_DECLARE_EXCHANGES_PROP_NAME, "true"));
+ private final boolean _bindQueues =
+ Boolean.parseBoolean(System.getProperty(ClientProperties.QPID_BIND_QUEUES_PROP_NAME, "true"));
+
private final boolean _useAMQPEncodedMapMessage;
private final boolean _useAMQPEncodedStreamMessage;
@@ -2870,10 +2873,13 @@ public abstract class AMQSession<C exten
{
declareQueue(amqd, consumer.isNoLocal(), nowait);
}
- if(!isBound(amqd.getExchangeName(), amqd.getAMQQueueName(), amqd.getRoutingKey()))
+ if (_bindQueues)
{
- bindQueue(amqd.getAMQQueueName(), amqd.getRoutingKey(),
- amqd instanceof AMQTopic ? consumer.getArguments() : null, amqd.getExchangeName(), amqd, nowait);
+ if(!isBound(amqd.getExchangeName(), amqd.getAMQQueueName(), amqd.getRoutingKey()))
+ {
+ bindQueue(amqd.getAMQQueueName(), amqd.getRoutingKey(),
+ amqd instanceof AMQTopic ? consumer.getArguments() : null, amqd.getExchangeName(), amqd, nowait);
+ }
}
}
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_8.java Tue Jan 7 17:58:41 2014
@@ -116,7 +116,7 @@ public class BasicMessageConsumer_0_8 ex
{
return getMessageFactory().createMessage(messageFrame.getDeliveryTag(),
- messageFrame.isRedelivered(), messageFrame.getExchange(),
+ messageFrame.isRedelivered(), messageFrame.getExchange() == null ? AMQShortString.EMPTY_STRING : messageFrame.getExchange(),
messageFrame.getRoutingKey(), messageFrame.getContentHeader(), messageFrame.getBodies(),
_queueDestinationCache, _topicDestinationCache);
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java Tue Jan 7 17:58:41 2014
@@ -60,7 +60,10 @@ public abstract class AbstractAMQMessage
_exchangeTypeToDestinationType.put(ExchangeDefaults.FANOUT_EXCHANGE_CLASS, AMQDestination.TOPIC_TYPE);
_exchangeTypeToDestinationType.put(ExchangeDefaults.HEADERS_EXCHANGE_CLASS, AMQDestination.QUEUE_TYPE);
- _exchangeMap.put("", new ExchangeInfo("","",AMQDestination.QUEUE_TYPE));
+ _exchangeMap.put(ExchangeDefaults.DEFAULT_EXCHANGE_NAME,
+ new ExchangeInfo(ExchangeDefaults.DEFAULT_EXCHANGE_NAME,
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS,
+ AMQDestination.QUEUE_TYPE));
_exchangeMap.put(ExchangeDefaults.DIRECT_EXCHANGE_NAME,
new ExchangeInfo(ExchangeDefaults.DIRECT_EXCHANGE_NAME,
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java Tue Jan 7 17:58:41 2014
@@ -230,6 +230,12 @@ public class ClientProperties
* producer/consumer creation when using BindingURLs.
*/
public static final String QPID_DECLARE_EXCHANGES_PROP_NAME = "qpid.declare_exchanges";
+ /**
+ * System property to control whether the client will bind queues during
+ * consumer creation when using BindingURLs.
+ */
+ public static final String QPID_BIND_QUEUES_PROP_NAME = "qpid.bind_queues";
+
public static final String VERIFY_QUEUE_ON_SEND = "qpid.verify_queue_on_send";
public static final String QPID_MAX_CACHED_ADDR_OPTION_STRINGS = "qpid.max_cached_address_option_strings";
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/message/JMSDestinationTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/message/JMSDestinationTest.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/message/JMSDestinationTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/client/message/JMSDestinationTest.java Tue Jan 7 17:58:41 2014
@@ -20,18 +20,17 @@
*/
package org.apache.qpid.test.client.message;
-import org.apache.qpid.client.AMQAnyDestination;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.CustomJMSXProperty;
import org.apache.qpid.configuration.ClientProperties;
-import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.management.common.mbeans.ManagedQueue;
import org.apache.qpid.test.utils.JMXTestUtils;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
import javax.jms.Connection;
import javax.jms.Destination;
+import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -41,7 +40,7 @@ import javax.jms.Session;
import javax.jms.Topic;
import javax.management.openmbean.CompositeDataSupport;
import javax.management.openmbean.TabularData;
-import java.nio.BufferOverflowException;
+
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -88,15 +87,15 @@ public class JMSDestinationTest extends
_connection.start();
- Message message = consumer.receive(10000);
+ Message receivedMessage = consumer.receive(10000);
- assertNotNull("Message should not be null", message);
+ assertNotNull("Message should not be null", receivedMessage);
- Destination destination = message.getJMSDestination();
+ Destination receivedDestination = receivedMessage.getJMSDestination();
- assertNotNull("JMSDestination should not be null", destination);
+ assertNotNull("JMSDestination should not be null", receivedDestination);
- assertEquals("Incorrect Destination type", queue.getClass(), destination.getClass());
+ assertEquals("Incorrect Destination type", queue.getClass(), receivedDestination.getClass());
}
/**
@@ -115,15 +114,14 @@ public class JMSDestinationTest extends
_connection.start();
- Message message = consumer.receive(10000);
-
- assertNotNull("Message should not be null", message);
+ Message receivedMessage = consumer.receive(10000);
- Destination destination = message.getJMSDestination();
+ assertNotNull("Message should not be null", receivedMessage);
- assertNotNull("JMSDestination should not be null", destination);
+ Destination receivedDestination = receivedMessage.getJMSDestination();
- assertEquals("Incorrect Destination type", topic.getClass(), destination.getClass());
+ assertNotNull("JMSDestination should not be null", receivedDestination);
+ assertEquals("Incorrect Destination type", topic.getClass(), receivedDestination.getClass());
}
/**
@@ -191,11 +189,11 @@ public class JMSDestinationTest extends
assertNotNull("Message should not be null", message);
- Destination destination = message.getJMSDestination();
+ Destination receivedDestination = message.getJMSDestination();
- assertNotNull("JMSDestination should not be null", destination);
+ assertNotNull("JMSDestination should not be null", receivedDestination);
- assertEquals("Incorrect Destination type", queue.getClass(), destination.getClass());
+ assertEquals("Incorrect Destination type", queue.getClass(), receivedDestination.getClass());
}
finally
@@ -238,11 +236,11 @@ public class JMSDestinationTest extends
assertNotNull("Message should not be null", _message);
- Destination destination = _message.getJMSDestination();
+ Destination receivedDestination = _message.getJMSDestination();
- assertNotNull("JMSDestination should not be null", destination);
+ assertNotNull("JMSDestination should not be null", receivedDestination);
- assertEquals("Incorrect Destination type", queue.getClass(), destination.getClass());
+ assertEquals("Incorrect Destination type", queue.getClass(), receivedDestination.getClass());
}
/**
@@ -305,17 +303,7 @@ public class JMSDestinationTest extends
// b) we can actually send without a BufferOverFlow.
MessageProducer producer = session08.createProducer(queue);
-
- try
- {
- producer.send(message);
- }
- catch (BufferOverflowException bofe)
- {
- // Print the stack trace so we can validate where the execption occured.
- bofe.printStackTrace();
- fail("BufferOverflowException thrown during send");
- }
+ producer.send(message);
message = consumer.receive(1000);
@@ -327,45 +315,45 @@ public class JMSDestinationTest extends
}
- /**
- * Send a message to a custom exchange and then verify
- * the message received has the proper destination set
- *
- * @throws Exception
- */
- public void testGetDestinationWithCustomExchange() throws Exception
+ public void testQueueWithBindingUrlUsingCustomExchange() throws Exception
{
-
- AMQDestination dest = new AMQAnyDestination(new AMQShortString("my-exchange"),
- new AMQShortString("direct"),
- new AMQShortString("test"),
- false,
- false,
- new AMQShortString("test"),
- false,
- new AMQShortString[]{new AMQShortString("test")});
-
- // to force the creation of my-exchange.
- sendMessage(_session, dest, 1);
-
- MessageProducer prod = _session.createProducer(dest);
+ String exchangeName = "exch_" + getTestQueueName();
+ String queueName = "queue_" + getTestQueueName();
+ String address = String.format("direct://%s/%s/%s?routingkey='%s'", exchangeName, queueName, queueName, queueName);
+ sendReceive(address);
+ }
+
+ public void testQueueWithBindingUrlUsingAmqDirectExchange() throws Exception
+ {
+ String queueName = getTestQueueName();
+ String address = String.format("direct://amq.direct/%s/%s?routingkey='%s'", queueName, queueName, queueName);
+ sendReceive(address);
+ }
+
+ public void testQueueWithBindingUrlUsingDefaultExchange() throws Exception
+ {
+ String queueName = getTestQueueName();
+ String address = String.format("direct:///%s/%s?routingkey='%s'", queueName, queueName, queueName);
+ sendReceive(address);
+ }
+
+ private void sendReceive(String address) throws JMSException, Exception
+ {
+ Destination dest = _session.createQueue(address);
MessageConsumer consumer = _session.createConsumer(dest);
-
+
_connection.start();
sendMessage(_session, dest, 1);
-
- Message message = consumer.receive(10000);
- assertNotNull("Message should not be null", message);
+ Message receivedMessage = consumer.receive(10000);
- Destination destination = message.getJMSDestination();
+ assertNotNull("Message should not be null", receivedMessage);
- assertNotNull("JMSDestination should not be null", destination);
+ Destination receivedDestination = receivedMessage.getJMSDestination();
- assertEquals("Incorrect Destination name", "my-exchange", dest.getExchangeName().asString());
- assertEquals("Incorrect Destination type", "direct", dest.getExchangeClass().asString());
- assertEquals("Incorrect Routing Key", "test", dest.getRoutingKey().asString());
+ assertNotNull("JMSDestination should not be null", receivedDestination);
+ assertEquals("JMSDestination should match that sent", address, receivedDestination.toString());
}
}
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/DynamicQueueExchangeCreateTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/DynamicQueueExchangeCreateTest.java?rev=1556292&r1=1556291&r2=1556292&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/DynamicQueueExchangeCreateTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/DynamicQueueExchangeCreateTest.java Tue Jan 7 17:58:41 2014
@@ -31,6 +31,7 @@ import org.apache.qpid.test.utils.QpidBr
import org.apache.qpid.url.BindingURL;
import javax.jms.Connection;
+import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.Session;
@@ -158,6 +159,28 @@ public class DynamicQueueExchangeCreateT
assertFalse("exchange should not exist", _jmxUtils.doesManagedObjectExist(exchangeObjectName2));
}
+ public void testQueueNotBoundDuringConsumerCreation() throws Exception
+ {
+ setSystemProperty(ClientProperties.QPID_BIND_QUEUES_PROP_NAME, "false");
+ setSystemProperty(ClientProperties.VERIFY_QUEUE_ON_SEND, "true");
+
+ Connection connection = getConnection();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ Queue queue = session.createQueue(getTestQueueName());
+ session.createConsumer(queue);
+
+ try
+ {
+ session.createProducer(queue).send(session.createMessage());
+ fail("JMSException should be thrown as the queue does not exist");
+ }
+ catch (InvalidDestinationException ide)
+ {
+ //PASS
+ }
+ }
private void checkExceptionErrorCode(JMSException original, AMQConstant code)
{
Exception linked = original.getLinkedException();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org