You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2007/02/17 00:11:44 UTC
svn commit: r508649 [1/2] - in /incubator/qpid/trunk/qpid/java: broker/etc/
broker/src/main/java/org/apache/qpid/server/configuration/
broker/src/main/java/org/apache/qpid/server/exchange/
broker/src/main/java/org/apache/qpid/server/handler/ broker/src...
Author: rgodfrey
Date: Fri Feb 16 15:11:41 2007
New Revision: 508649
URL: http://svn.apache.org/viewvc?view=rev&rev=508649
Log:
QPID-375 : remove assumptions on standard exchanges (amq.direct, amq.topic, etc), allow other exchanges to be created through virtualhosts.xml
Modified:
incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryQueue.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryTopic.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/Session.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/cluster/Client.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/flow/ChannelFlowTest.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/latency/LatencyTest.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/multiconsumer/AMQTest.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestSubscriber.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/topic/MessageFactory.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Ping.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Pong.java
incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Start.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/BytesMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/ObjectMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PropertyValueTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/PubSubTwoConnectionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/ReceiveTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SelectorTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/SessionStartTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/AMQConnectionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/AMQSessionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/CloseWithBlockingReceiveTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionStartTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connection/ConnectionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/connectionurl/ConnectionURLTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/destinationurl/DestinationURLTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Client.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Service.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/SpecialQueue.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/close/TopicPublisherCloseTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSDestinationTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/JMSPropertiesTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/MessageConverterTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicPublisherTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/testutil/Config.java
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongBouncer.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Listener.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/topic/MessageFactory.java
Modified: incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml (original)
+++ incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml Fri Feb 16 15:11:41 2007
@@ -24,58 +24,54 @@
<virtualhost>
<name>localhost</name>
<localhost>
- <minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
- <maximumMessageCount>5000</maximumMessageCount>
- <queue>
- <name>queue</name>
+ <exchanges>
+ <exchange>
+ <type>direct</type>
+ <name>test.direct</name>
+ <durable>true</durable>
+ </exchange>
+ <exchange>
+ <type>topic</type>
+ <name>test.topic</name>
+ </exchange>
+ </exchanges>
+ <queues>
+ <exchange>amq.direct</exchange>
+ <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
+ <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
+ <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
+
+ <queue>
+ <name>queue</name>
+ </queue>
+ <queue>
+ <name>ping</name>
+ </queue>
<queue>
- <exchange>amq.direct</exchange>
- <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
- <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
- <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
- </queue>
- </queue>
- <queue>
- <name>ping</name>
- <ping>
- <exchange>amq.direct</exchange>
- <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
- <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
- <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
- </ping>
- </queue>
+ <name>test-queue</name>
+ <test-queue>
+ <exchange>test.direct</exchange>
+ <durable>true</durable>
+ </test-queue>
+ </queue>
+ <queue>
+ <name>test-ping</name>
+ <test-ping>
+ <exchange>test.direct</exchange>
+ </test-ping>
+ </queue>
+
+ </queues>
</localhost>
</virtualhost>
+
+
<virtualhost>
<name>development</name>
<development>
<minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
<maximumMessageCount>5000</maximumMessageCount>
- <queue>
- <name>queue</name>
- <queue>
- <exchange>amq.direct</exchange>
- <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
- <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
- <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
- </queue>
- </queue>
- <queue>
- <name>ping</name>
- <ping>
- <exchange>amq.direct</exchange>
- <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
- <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
- <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
- </ping>
- </queue>
- </development>
- </virtualhost>
- <virtualhost>
- <name>test</name>
- <test>
- <minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
- <maximumMessageCount>5000</maximumMessageCount>
+ <queues>
<queue>
<name>queue</name>
<queue>
@@ -94,6 +90,34 @@
<maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
</ping>
</queue>
+ </queues>
+ </development>
+ </virtualhost>
+ <virtualhost>
+ <name>test</name>
+ <test>
+ <minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
+ <maximumMessageCount>5000</maximumMessageCount>
+ <queues>
+ <queue>
+ <name>queue</name>
+ <queue>
+ <exchange>amq.direct</exchange>
+ <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
+ <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
+ <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
+ </queue>
+ </queue>
+ <queue>
+ <name>ping</name>
+ <ping>
+ <exchange>amq.direct</exchange>
+ <maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
+ <maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
+ <maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
+ </ping>
+ </queue>
+ </queues>
</test>
</virtualhost>
</virtualhosts>
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java Fri Feb 16 15:11:41 2007
@@ -32,6 +32,7 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.exchange.ExchangeFactory;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -71,7 +72,16 @@
throw new ConfigurationException("Unknown virtual host: " + virtualHostName);
}
- List queueNames = configuration.getList("queue.name");
+ List exchangeNames = configuration.getList("exchanges.exchange.name");
+
+ for(Object exchangeNameObj : exchangeNames)
+ {
+ String exchangeName = String.valueOf(exchangeNameObj);
+ configureExchange(virtualHost, exchangeName, configuration);
+ }
+
+
+ List queueNames = configuration.getList("queues.queue.name");
for(Object queueNameObj : queueNames)
{
@@ -81,12 +91,49 @@
}
+ private void configureExchange(VirtualHost virtualHost, String exchangeNameString, Configuration configuration) throws AMQException
+ {
+
+ CompositeConfiguration exchangeConfiguration = new CompositeConfiguration();
+
+ exchangeConfiguration.addConfiguration(configuration.subset("exchanges.exchange."+ exchangeNameString));
+ exchangeConfiguration.addConfiguration(configuration.subset("exchanges"));
+
+ QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
+ MessageStore messageStore = virtualHost.getMessageStore();
+ ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry();
+ ExchangeFactory exchangeFactory = virtualHost.getExchangeFactory();
+
+ AMQShortString exchangeName = new AMQShortString(exchangeNameString);
+
+
+ Exchange exchange;
+
+
+
+ synchronized (exchangeRegistry)
+ {
+ exchange = exchangeRegistry.getExchange(exchangeName);
+ if(exchange == null)
+ {
+
+ AMQShortString type = new AMQShortString(exchangeConfiguration.getString("type","direct"));
+ boolean durable = exchangeConfiguration.getBoolean("durable",false);
+ boolean autodelete = exchangeConfiguration.getBoolean("autodelete",false);
+
+ Exchange newExchange = exchangeFactory.createExchange(exchangeName,type,durable,autodelete,0);
+ exchangeRegistry.registerExchange(newExchange);
+ }
+
+ }
+ }
+
private void configureQueue(VirtualHost virtualHost, String queueNameString, Configuration configuration) throws AMQException, ConfigurationException
{
CompositeConfiguration queueConfiguration = new CompositeConfiguration();
- queueConfiguration.addConfiguration(configuration.subset("queue."+ queueNameString));
- queueConfiguration.addConfiguration(configuration);
+ queueConfiguration.addConfiguration(configuration.subset("queues.queue."+ queueNameString));
+ queueConfiguration.addConfiguration(configuration.subset("queues"));
QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
MessageStore messageStore = virtualHost.getMessageStore();
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java Fri Feb 16 15:11:41 2007
@@ -55,10 +55,6 @@
public void registerExchange(Exchange exchange)
{
- if(_defaultExchange == null)
- {
- setDefaultExchange(exchange);
- }
_exchangeMap.put(exchange.getName(), exchange);
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java Fri Feb 16 15:11:41 2007
@@ -63,7 +63,7 @@
// TODO: check the delivery tag field details - is it unique across the broker or per subscriber?
if (body.exchange == null)
{
- body.exchange = ExchangeDefaults.DIRECT_EXCHANGE_NAME;
+ body.exchange = ExchangeDefaults.DEFAULT_EXCHANGE_NAME;
}
VirtualHost vHost = session.getVirtualHost();
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java Fri Feb 16 15:11:41 2007
@@ -107,7 +107,7 @@
queueRegistry.registerQueue(queue);
if (autoRegister)
{
- Exchange defaultExchange = exchangeRegistry.getExchange(ExchangeDefaults.DIRECT_EXCHANGE_NAME);
+ Exchange defaultExchange = exchangeRegistry.getDefaultExchange();
defaultExchange.registerQueue(body.queue, queue, null);
queue.bind(body.queue, defaultExchange);
_log.info("Queue " + body.queue + " bound to default exchange");
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ExchangeInitialiser.java Fri Feb 16 15:11:41 2007
@@ -28,13 +28,14 @@
public class ExchangeInitialiser
{
- public void initialise(ExchangeFactory factory, ExchangeRegistry registry) throws AMQException{
+ public void initialise(ExchangeFactory factory, ExchangeRegistry registry) throws AMQException{
+ define(registry, factory, ExchangeDefaults.DEFAULT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
define(registry, factory, ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
define(registry, factory, ExchangeDefaults.TOPIC_EXCHANGE_NAME, ExchangeDefaults.TOPIC_EXCHANGE_CLASS);
define(registry, factory, ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS);
define(registry, factory, ExchangeDefaults.FANOUT_EXCHANGE_NAME, ExchangeDefaults.FANOUT_EXCHANGE_CLASS);
- registry.setDefaultExchange(registry.getExchange(ExchangeDefaults.DIRECT_EXCHANGE_NAME));
+ registry.setDefaultExchange(registry.getExchange(ExchangeDefaults.DEFAULT_EXCHANGE_NAME));
}
private void define(ExchangeRegistry r, ExchangeFactory f,
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java Fri Feb 16 15:11:41 2007
@@ -20,37 +20,6 @@
*/
package org.apache.qpid.client;
-import java.io.IOException;
-import java.net.ConnectException;
-import java.nio.channels.UnresolvedAddressException;
-import java.text.MessageFormat;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import javax.jms.ConnectionConsumer;
-import javax.jms.ConnectionMetaData;
-import javax.jms.Destination;
-import javax.jms.ExceptionListener;
-import javax.jms.IllegalStateException;
-import javax.jms.JMSException;
-import javax.jms.Queue;
-import javax.jms.QueueConnection;
-import javax.jms.QueueSession;
-import javax.jms.ServerSessionPool;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.jms.TopicConnection;
-import javax.jms.TopicSession;
-import javax.naming.NamingException;
-import javax.naming.Reference;
-import javax.naming.Referenceable;
-import javax.naming.StringRefAddr;
-
-import org.apache.log4j.Logger;
import org.apache.qpid.AMQConnectionFailureException;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQUndeliveredException;
@@ -59,6 +28,8 @@
import org.apache.qpid.client.protocol.AMQProtocolHandler;
import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicQosBody;
import org.apache.qpid.framing.BasicQosOkBody;
import org.apache.qpid.framing.ChannelOpenBody;
@@ -73,6 +44,25 @@
import org.apache.qpid.jms.FailoverPolicy;
import org.apache.qpid.url.URLSyntaxException;
+import org.apache.log4j.Logger;
+
+import javax.jms.*;
+import javax.jms.IllegalStateException;
+import javax.naming.NamingException;
+import javax.naming.Reference;
+import javax.naming.Referenceable;
+import javax.naming.StringRefAddr;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.nio.channels.UnresolvedAddressException;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
public class AMQConnection extends Closeable implements Connection, QueueConnection, TopicConnection, Referenceable
{
private static final Logger _logger = Logger.getLogger(AMQConnection.class);
@@ -157,12 +147,17 @@
* The connection meta data
*/
private QpidConnectionMetaData _connectionMetaData;
-
+
/**
* Configuration info for SSL
*/
private SSLConfiguration _sslConfiguration;
+ private AMQShortString _defaultTopicExchangeName = ExchangeDefaults.TOPIC_EXCHANGE_NAME;
+ private AMQShortString _defaultQueueExchangeName = ExchangeDefaults.DIRECT_EXCHANGE_NAME;
+ private AMQShortString _temporaryTopicExchangeName = ExchangeDefaults.TOPIC_EXCHANGE_NAME;
+ private AMQShortString _temporaryQueueExchangeName = ExchangeDefaults.DIRECT_EXCHANGE_NAME;
+
/**
* @param broker brokerdetails
* @param username username
@@ -180,7 +175,7 @@
(clientName == null ? "" : clientName) + "/" +
virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'"), null);
}
-
+
/**
* @param broker brokerdetails
* @param username username
@@ -198,20 +193,20 @@
(clientName == null ? "" : clientName) + "/" +
virtualHost + "?brokerlist='" + AMQBrokerDetails.checkTransport(broker) + "'"), sslConfig);
}
-
+
public AMQConnection(String host, int port, String username, String password,
String clientName, String virtualHost) throws AMQException, URLSyntaxException
{
this(host, port, false, username, password, clientName, virtualHost, null);
}
-
+
public AMQConnection(String host, int port, String username, String password,
- String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
+ String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
{
- this(host, port, false, username, password, clientName, virtualHost, sslConfig);
+ this(host, port, false, username, password, clientName, virtualHost, sslConfig);
}
-
+
public AMQConnection(String host, int port, boolean useSSL, String username, String password,
String clientName, String virtualHost, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
@@ -234,12 +229,12 @@
{
this(new AMQConnectionURL(connection), null);
}
-
+
public AMQConnection(String connection, SSLConfiguration sslConfig) throws AMQException, URLSyntaxException
{
this(new AMQConnectionURL(connection), sslConfig);
}
-
+
public AMQConnection(ConnectionURL connectionURL, SSLConfiguration sslConfig) throws AMQException
{
@@ -257,6 +252,28 @@
_password = connectionURL.getPassword();
setVirtualHost(connectionURL.getVirtualHost());
+
+ if (connectionURL.getDefaultQueueExchangeName() != null)
+ {
+ _defaultQueueExchangeName = connectionURL.getDefaultQueueExchangeName();
+ }
+
+ if (connectionURL.getDefaultTopicExchangeName() != null)
+ {
+ _defaultTopicExchangeName = connectionURL.getDefaultTopicExchangeName();
+ }
+
+ if (connectionURL.getTemporaryQueueExchangeName() != null)
+ {
+ _temporaryQueueExchangeName = connectionURL.getTemporaryQueueExchangeName();
+ }
+
+ if (connectionURL.getTemporaryTopicExchangeName() != null)
+ {
+ _temporaryTopicExchangeName = connectionURL.getTemporaryTopicExchangeName();
+ }
+
+
_failoverPolicy = new FailoverPolicy(connectionURL);
_protocolHandler = new AMQProtocolHandler(this);
@@ -440,7 +457,7 @@
}
}
- public Session createSession(final boolean transacted, final int acknowledgeMode) throws JMSException
+ public org.apache.qpid.jms.Session createSession(final boolean transacted, final int acknowledgeMode) throws JMSException
{
return createSession(transacted, acknowledgeMode, AMQSession.DEFAULT_PREFETCH_HIGH_MARK);
}
@@ -1070,9 +1087,53 @@
AMQConnectionFactory.class.getName(),
null); // factory location
}
-
+
public SSLConfiguration getSSLConfiguration()
{
- return _sslConfiguration;
+ return _sslConfiguration;
+ }
+
+ public AMQShortString getDefaultTopicExchangeName()
+ {
+ return _defaultTopicExchangeName;
+ }
+
+
+ public void setDefaultTopicExchangeName(AMQShortString defaultTopicExchangeName)
+ {
+ _defaultTopicExchangeName = defaultTopicExchangeName;
+ }
+
+
+ public AMQShortString getDefaultQueueExchangeName()
+ {
+ return _defaultQueueExchangeName;
+ }
+
+
+ public void setDefaultQueueExchangeName(AMQShortString defaultQueueExchangeName)
+ {
+ _defaultQueueExchangeName = defaultQueueExchangeName;
+ }
+
+ public AMQShortString getTemporaryTopicExchangeName()
+ {
+ return _temporaryTopicExchangeName;
+ }
+
+ public AMQShortString getTemporaryQueueExchangeName()
+ {
+ return _temporaryQueueExchangeName; //To change body of created methods use File | Settings | File Templates.
+ }
+
+
+ public void setTemporaryTopicExchangeName(AMQShortString temporaryTopicExchangeName)
+ {
+ _temporaryTopicExchangeName = temporaryTopicExchangeName;
+ }
+
+ public void setTemporaryQueueExchangeName(AMQShortString temporaryQueueExchangeName)
+ {
+ _temporaryQueueExchangeName = temporaryQueueExchangeName;
}
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionURL.java Fri Feb 16 15:11:41 2007
@@ -20,6 +20,12 @@
*/
package org.apache.qpid.client;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.BrokerDetails;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.url.URLHelper;
+import org.apache.qpid.url.URLSyntaxException;
+
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
@@ -27,11 +33,6 @@
import java.util.List;
import java.util.StringTokenizer;
-import org.apache.qpid.jms.BrokerDetails;
-import org.apache.qpid.jms.ConnectionURL;
-import org.apache.qpid.url.URLHelper;
-import org.apache.qpid.url.URLSyntaxException;
-
public class AMQConnectionURL implements ConnectionURL
{
private String _url;
@@ -43,6 +44,11 @@
private String _username;
private String _password;
private String _virtualHost;
+ private AMQShortString _defaultQueueExchangeName;
+ private AMQShortString _defaultTopicExchangeName;
+ private AMQShortString _temporaryTopicExchangeName;
+ private AMQShortString _temporaryQueueExchangeName;
+
public AMQConnectionURL(String fullURL) throws URLSyntaxException
{
@@ -107,7 +113,7 @@
if (userInfo == null)
{
throw URLHelper.parseError(AMQ_PROTOCOL.length() + 3,
- "User information not found on url", fullURL);
+ "User information not found on url", fullURL);
}
else
{
@@ -161,7 +167,9 @@
{
if (slash != 0 && fullURL.charAt(slash - 1) == ':')
{
- throw URLHelper.parseError(slash - 2, fullURL.indexOf('?') - slash + 2, "Virtual host looks like a windows path, forward slash not allowed in URL", fullURL);
+ throw URLHelper.parseError(slash - 2, fullURL.indexOf('?') - slash + 2,
+ "Virtual host looks like a windows path, forward slash not allowed in URL",
+ fullURL);
}
else
{
@@ -181,7 +189,7 @@
if (colonIndex == -1)
{
throw URLHelper.parseError(AMQ_PROTOCOL.length() + 3, userinfo.length(),
- "Null password in user information not allowed.", _url);
+ "Null password in user information not allowed.", _url);
}
else
{
@@ -230,6 +238,29 @@
_options.remove(OPTIONS_FAILOVER);
}
+
+ if (_options.containsKey(OPTIONS_DEFAULT_TOPIC_EXCHANGE))
+ {
+ _defaultTopicExchangeName = new AMQShortString(_options.get(OPTIONS_DEFAULT_TOPIC_EXCHANGE));
+ }
+
+
+ if (_options.containsKey(OPTIONS_DEFAULT_QUEUE_EXCHANGE))
+ {
+ _defaultQueueExchangeName = new AMQShortString(_options.get(OPTIONS_DEFAULT_QUEUE_EXCHANGE));
+ }
+
+
+ if (_options.containsKey(OPTIONS_TEMPORARY_QUEUE_EXCHANGE))
+ {
+ _temporaryQueueExchangeName = new AMQShortString(_options.get(OPTIONS_TEMPORARY_QUEUE_EXCHANGE));
+ }
+
+
+ if (_options.containsKey(OPTIONS_TEMPORARY_TOPIC_EXCHANGE))
+ {
+ _temporaryTopicExchangeName = new AMQShortString(_options.get(OPTIONS_TEMPORARY_TOPIC_EXCHANGE));
+ }
}
public String getURL()
@@ -330,6 +361,26 @@
public void setOption(String key, String value)
{
_options.put(key, value);
+ }
+
+ public AMQShortString getDefaultQueueExchangeName()
+ {
+ return _defaultQueueExchangeName;
+ }
+
+ public AMQShortString getDefaultTopicExchangeName()
+ {
+ return _defaultTopicExchangeName;
+ }
+
+ public AMQShortString getTemporaryQueueExchangeName()
+ {
+ return _temporaryQueueExchangeName;
+ }
+
+ public AMQShortString getTemporaryTopicExchangeName()
+ {
+ return _temporaryTopicExchangeName;
}
public String toString()
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQDestination.java Fri Feb 16 15:11:41 2007
@@ -145,12 +145,12 @@
public boolean isTopic()
{
- return ExchangeDefaults.TOPIC_EXCHANGE_NAME.equals(_exchangeName);
+ return ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(_exchangeClass);
}
public boolean isQueue()
{
- return ExchangeDefaults.DIRECT_EXCHANGE_NAME.equals(_exchangeName);
+ return ExchangeDefaults.DIRECT_EXCHANGE_CLASS.equals(_exchangeClass);
}
public AMQShortString getDestinationName()
@@ -411,11 +411,11 @@
if (exchangeClass.equals(ExchangeDefaults.DIRECT_EXCHANGE_CLASS))
{
- return new AMQQueue(destinationName,queueName,isExclusive,isAutoDelete,isDurable);
+ return new AMQQueue(exchangeName,destinationName,queueName,isExclusive,isAutoDelete,isDurable);
}
else if (exchangeClass.equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS))
{
- return new AMQTopic(destinationName,isAutoDelete,queueName,isDurable);
+ return new AMQTopic(exchangeName,destinationName,isAutoDelete,queueName,isDurable);
}
else if (exchangeClass.equals(ExchangeDefaults.HEADERS_EXCHANGE_CLASS))
{
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueue.java Fri Feb 16 15:11:41 2007
@@ -21,6 +21,7 @@
package org.apache.qpid.client;
import javax.jms.Queue;
+import javax.jms.Connection;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
@@ -43,9 +44,19 @@
* Create a reference to a non temporary queue. Note this does not actually imply the queue exists.
* @param name the name of the queue
*/
- public AMQQueue(AMQShortString name)
+ public AMQQueue(AMQShortString exchangeName, String name)
{
- this(name, false);
+ this(exchangeName, new AMQShortString(name));
+ }
+
+
+ /**
+ * Create a reference to a non temporary queue. Note this does not actually imply the queue exists.
+ * @param name the name of the queue
+ */
+ public AMQQueue(AMQShortString exchangeName, AMQShortString name)
+ {
+ this(exchangeName, name, false);
}
public AMQQueue(AMQShortString exchangeName, AMQShortString routingKey, AMQShortString queueName)
@@ -58,9 +69,20 @@
* Create a reference to a non temporary queue. Note this does not actually imply the queue exists.
* @param name the name of the queue
*/
- public AMQQueue(String name)
+ public AMQQueue(String exchangeName, String name)
{
- this(new AMQShortString(name), false);
+ this(new AMQShortString(exchangeName), new AMQShortString(name), false);
+ }
+
+
+ public AMQQueue(AMQConnection connection, String name)
+ {
+ this(connection.getDefaultQueueExchangeName(),name);
+ }
+
+ public AMQQueue(AMQConnection connection, String name, boolean temporary)
+ {
+ this(connection.getDefaultQueueExchangeName(), new AMQShortString(name),temporary);
}
@@ -71,9 +93,9 @@
* @param temporary if true the broker will generate a queue name, also if true then the queue is autodeleted
* and exclusive
*/
- public AMQQueue(String name, boolean temporary)
+ public AMQQueue(String exchangeName, String name, boolean temporary)
{
- this(new AMQShortString(name),temporary);
+ this(new AMQShortString(exchangeName), new AMQShortString(name),temporary);
}
@@ -84,11 +106,11 @@
* @param temporary if true the broker will generate a queue name, also if true then the queue is autodeleted
* and exclusive
*/
- public AMQQueue(AMQShortString name, boolean temporary)
+ public AMQQueue(AMQShortString exchangeName, AMQShortString name, boolean temporary)
{
// queue name is set to null indicating that the broker assigns a name in the case of temporary queues
// temporary queues are typically used as response queues
- this(name, temporary?null:name, temporary, temporary, !temporary);
+ this(exchangeName, name, temporary?null:name, temporary, temporary, !temporary);
}
@@ -99,19 +121,20 @@
* @param exclusive true if the queue should only permit a single consumer
* @param autoDelete true if the queue should be deleted automatically when the last consumers detaches
*/
- public AMQQueue(AMQShortString destinationName, AMQShortString queueName, boolean exclusive, boolean autoDelete)
+ public AMQQueue(AMQShortString exchangeName, AMQShortString destinationName, AMQShortString queueName, boolean exclusive, boolean autoDelete)
{
- this(destinationName, queueName, exclusive, autoDelete, false);
+ this(exchangeName, destinationName, queueName, exclusive, autoDelete, false);
}
- public AMQQueue(AMQShortString destinationName, AMQShortString queueName, boolean exclusive, boolean autoDelete, boolean durable)
+ public AMQQueue(AMQShortString exchangeName, AMQShortString destinationName, AMQShortString queueName, boolean exclusive, boolean autoDelete, boolean durable)
{
- super(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS, destinationName, exclusive,
+ super(exchangeName, ExchangeDefaults.DIRECT_EXCHANGE_CLASS, destinationName, exclusive,
autoDelete, queueName, durable);
}
-
+
+
public AMQShortString getRoutingKey()
{
return getAMQQueueName();
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Fri Feb 16 15:11:41 2007
@@ -1312,7 +1312,7 @@
checkNotClosed();
if (queueName.indexOf('/') == -1)
{
- return new AMQQueue(queueName);
+ return new AMQQueue(getDefaultQueueExchangeName(), new AMQShortString(queueName));
}
else
{
@@ -1330,6 +1330,11 @@
}
}
+ public AMQShortString getDefaultQueueExchangeName()
+ {
+ return _connection.getDefaultQueueExchangeName();
+ }
+
/**
* Creates a QueueReceiver wrapping a MessageConsumer
*
@@ -1379,7 +1384,7 @@
if (topicName.indexOf('/') == -1)
{
- return new AMQTopic(new AMQShortString(topicName));
+ return new AMQTopic(getDefaultTopicExchangeName(),new AMQShortString(topicName));
}
else
{
@@ -1397,6 +1402,11 @@
}
}
+ public AMQShortString getDefaultTopicExchangeName()
+ {
+ return _connection.getDefaultTopicExchangeName();
+ }
+
/**
* Creates a non-durable subscriber
*
@@ -1409,8 +1419,8 @@
public TopicSubscriber createSubscriber(Topic topic) throws JMSException
{
checkNotClosed();
- checkValidTopic(topic);
- AMQTopic dest = new AMQTopic(topic.getTopicName());
+ AMQTopic dest = checkValidTopic(topic);
+ //AMQTopic dest = new AMQTopic(topic.getTopicName());
return new TopicSubscriberAdaptor(dest, (BasicMessageConsumer) createConsumer(dest));
}
@@ -1428,16 +1438,16 @@
public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
{
checkNotClosed();
- checkValidTopic(topic);
- AMQTopic dest = new AMQTopic(topic.getTopicName());
+ AMQTopic dest = checkValidTopic(topic);
+ //AMQTopic dest = new AMQTopic(topic.getTopicName());
return new TopicSubscriberAdaptor(dest, (BasicMessageConsumer) createConsumer(dest, messageSelector, noLocal));
}
public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException
{
checkNotClosed();
- checkValidTopic(topic);
- AMQTopic dest = AMQTopic.createDurableTopic((AMQTopic) topic, name, _connection);
+ AMQTopic origTopic = checkValidTopic(topic);
+ AMQTopic dest = AMQTopic.createDurableTopic(origTopic, name, _connection);
TopicSubscriberAdaptor subscriber = _subscriptions.get(name);
if (subscriber != null)
{
@@ -1464,8 +1474,8 @@
}
// if the queue is bound to the exchange but NOT for this topic, then the JMS spec
// says we must trash the subscription.
- if (isQueueBound(dest.getAMQQueueName()) &&
- !isQueueBound(dest.getAMQQueueName(), topicName))
+ if (isQueueBound(dest.getExchangeName(),dest.getAMQQueueName()) &&
+ !isQueueBound(dest.getExchangeName(),dest.getAMQQueueName(), topicName))
{
deleteQueue(dest.getAMQQueueName());
}
@@ -1556,7 +1566,7 @@
}
else
{
- if (isQueueBound(AMQTopic.getDurableTopicQueueName(name, _connection)))
+ if (isQueueBound(getDefaultTopicExchangeName(), AMQTopic.getDurableTopicQueueName(name, _connection)))
{
deleteQueue(AMQTopic.getDurableTopicQueueName(name, _connection));
}
@@ -1567,17 +1577,17 @@
}
}
- boolean isQueueBound(AMQShortString queueName) throws JMSException
+ boolean isQueueBound(AMQShortString exchangeName, AMQShortString queueName) throws JMSException
{
- return isQueueBound(queueName, null);
+ return isQueueBound(exchangeName, queueName, null);
}
- boolean isQueueBound(AMQShortString queueName, AMQShortString routingKey) throws JMSException
+ boolean isQueueBound(AMQShortString exchangeName, AMQShortString queueName, AMQShortString routingKey) throws JMSException
{
// TODO: Be aware of possible changes to parameter order as versions change.
AMQFrame boundFrame = ExchangeBoundBody.createAMQFrame(_channelId,
getProtocolMajorVersion(), getProtocolMinorVersion(), // AMQP version (major, minor)
- ExchangeDefaults.TOPIC_EXCHANGE_NAME, // exchange
+ exchangeName, // exchange
queueName, // queue
routingKey); // routingKey
AMQMethodEvent response = null;
@@ -1858,7 +1868,7 @@
/*
* I could have combined the last 3 methods, but this way it improves readability
*/
- private void checkValidTopic(Topic topic) throws JMSException
+ private AMQTopic checkValidTopic(Topic topic) throws JMSException
{
if (topic == null)
{
@@ -1866,8 +1876,13 @@
}
if ((topic instanceof TemporaryDestination) && ((TemporaryDestination) topic).getSession() != this)
{
- throw new JMSException("Cannot create a subscription on a temporary topic created in another session");
+ throw new javax.jms.InvalidDestinationException("Cannot create a subscription on a temporary topic created in another session");
}
+ if(!(topic instanceof AMQTopic))
+ {
+ throw new javax.jms.InvalidDestinationException("Cannot create a subscription on topic created for another JMS Provider, class of topic provided is: " + topic.getClass().getName());
+ }
+ return (AMQTopic) topic;
}
private void checkValidQueue(Queue queue) throws InvalidDestinationException
@@ -1885,6 +1900,18 @@
throw new javax.jms.InvalidDestinationException("Invalid Queue");
}
}
+
+
+ public AMQShortString getTemporaryTopicExchangeName()
+ {
+ return _connection.getTemporaryTopicExchangeName();
+ }
+
+ public AMQShortString getTemporaryQueueExchangeName()
+ {
+ return _connection.getTemporaryQueueExchangeName();
+ }
+
public int getTicket()
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryQueue.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryQueue.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryQueue.java Fri Feb 16 15:11:41 2007
@@ -40,7 +40,7 @@
*/
public AMQTemporaryQueue(AMQSession session)
{
- super(new AMQShortString("TempQueue" + Long.toString(System.currentTimeMillis())), true);
+ super(session.getTemporaryQueueExchangeName(),new AMQShortString("TempQueue" + Long.toString(System.currentTimeMillis())), true);
_session = session;
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryTopic.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryTopic.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryTopic.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTemporaryTopic.java Fri Feb 16 15:11:41 2007
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.client;
+import org.apache.qpid.framing.AMQShortString;
+
import javax.jms.JMSException;
import javax.jms.TemporaryTopic;
@@ -36,7 +38,7 @@
*/
public AMQTemporaryTopic(AMQSession session)
{
- super("TempQueue" + Long.toString(System.currentTimeMillis()));
+ super(session.getTemporaryTopicExchangeName(),new AMQShortString("TempQueue" + Long.toString(System.currentTimeMillis())));
_session = session;
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQTopic.java Fri Feb 16 15:11:41 2007
@@ -39,32 +39,43 @@
super(binding);
}
- public AMQTopic(String name)
- {
- this(new AMQShortString(name));
- }
+// public AMQTopic(String exchangeName, String routingKey)
+// {
+// this(new AMQShortString(exchangeName), new AMQShortString(routingKey));
+// }
public AMQTopic(AMQShortString exchange, AMQShortString routingKey, AMQShortString queueName)
{
super(exchange, ExchangeDefaults.TOPIC_EXCHANGE_CLASS, routingKey, true, true, queueName, false);
}
+ public AMQTopic(AMQConnection conn, String routingKey)
+ {
+ this(conn.getDefaultTopicExchangeName(), new AMQShortString(routingKey));
+ }
+
+
+ public AMQTopic(AMQShortString exchangeName, String routingKey)
+ {
+ this(exchangeName, new AMQShortString(routingKey));
+ }
- public AMQTopic(AMQShortString name)
+ public AMQTopic(AMQShortString exchangeName, AMQShortString routingKey)
{
- this(name, true, null, false);
+ this(exchangeName, routingKey, null);
}
- public AMQTopic(AMQShortString name, boolean isAutoDelete, AMQShortString queueName, boolean isDurable)
+ public AMQTopic(AMQShortString exchangeName, AMQShortString name, boolean isAutoDelete, AMQShortString queueName, boolean isDurable)
{
- super(ExchangeDefaults.TOPIC_EXCHANGE_NAME, ExchangeDefaults.TOPIC_EXCHANGE_CLASS, name, true, isAutoDelete,
+ super(exchangeName, ExchangeDefaults.TOPIC_EXCHANGE_CLASS, name, true, isAutoDelete,
queueName, isDurable);
}
public static AMQTopic createDurableTopic(AMQTopic topic, String subscriptionName, AMQConnection connection)
throws JMSException
{
- return new AMQTopic(topic.getDestinationName(), false, getDurableTopicQueueName(subscriptionName, connection),
+ return new AMQTopic(topic.getExchangeName(), topic.getDestinationName(), false,
+ getDurableTopicQueueName(subscriptionName, connection),
true);
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/JMSBytesMessage.java Fri Feb 16 15:11:41 2007
@@ -380,4 +380,9 @@
}
}
+ public String toString()
+ {
+ return String.valueOf(System.identityHashCode(this));
+ }
+
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/ConnectionURL.java Fri Feb 16 15:11:41 2007
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.jms;
+import org.apache.qpid.framing.AMQShortString;
+
import java.util.List;
/**
@@ -35,6 +37,10 @@
public static final String OPTIONS_FAILOVER = "failover";
public static final String OPTIONS_FAILOVER_CYCLE = "cyclecount";
public static final String OPTIONS_SSL = "ssl";
+ public static final String OPTIONS_DEFAULT_TOPIC_EXCHANGE = "defaultTopicExchange";
+ public static final String OPTIONS_DEFAULT_QUEUE_EXCHANGE = "defaultQueueExchange";
+ public static final String OPTIONS_TEMPORARY_TOPIC_EXCHANGE = "temporaryTopicExchange";
+ public static final String OPTIONS_TEMPORARY_QUEUE_EXCHANGE = "temporaryQueueExchange";
String getURL();
@@ -69,4 +75,12 @@
String getOption(String key);
void setOption(String key, String value);
+
+ AMQShortString getDefaultQueueExchangeName();
+
+ AMQShortString getDefaultTopicExchangeName();
+
+ AMQShortString getTemporaryQueueExchangeName();
+
+ AMQShortString getTemporaryTopicExchangeName();
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/Session.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/Session.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/Session.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/Session.java Fri Feb 16 15:11:41 2007
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.jms;
+import org.apache.qpid.framing.AMQShortString;
+
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
@@ -88,4 +90,12 @@
*/
MessageProducer createProducer(Destination destination, boolean immediate)
throws JMSException;
+
+ AMQShortString getTemporaryTopicExchangeName();
+
+ AMQShortString getDefaultQueueExchangeName();
+
+ AMQShortString getDefaultTopicExchangeName();
+
+ AMQShortString getTemporaryQueueExchangeName();
}
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java Fri Feb 16 15:11:41 2007
@@ -47,6 +47,7 @@
import org.apache.qpid.url.AMQBindingURL;
import org.apache.qpid.url.BindingURL;
import org.apache.qpid.url.URLSyntaxException;
+import org.apache.qpid.exchange.ExchangeDefaults;
public class PropertiesFileInitialContextFactory implements InitialContextFactory
{
@@ -236,12 +237,12 @@
{
if(value instanceof AMQShortString)
{
- return new AMQQueue((AMQShortString) value);
+ return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, (AMQShortString) value);
}
else if (value instanceof String)
{
- return new AMQQueue(new AMQShortString((String) value));
+ return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
@@ -259,11 +260,11 @@
{
if(value instanceof AMQShortString)
{
- return new AMQTopic((AMQShortString)value);
+ return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, (AMQShortString)value);
}
else if (value instanceof String)
{
- return new AMQTopic(new AMQShortString((String) value));
+ return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, new AMQShortString((String) value));
}
else if (value instanceof BindingURL)
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/cluster/Client.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/cluster/Client.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/cluster/Client.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/cluster/Client.java Fri Feb 16 15:11:41 2007
@@ -25,6 +25,8 @@
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.url.URLSyntaxException;
import javax.jms.MessageListener;
@@ -48,8 +50,8 @@
this.name = name;
session = connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);
- AMQTopic topic = new AMQTopic("cluster_test_topic");
- AMQQueue queue = new AMQQueue("cluster_test_queue");
+ AMQTopic topic = new AMQTopic(((AMQSession)session).getDefaultTopicExchangeName(), new AMQShortString("cluster_test_topic"));
+ AMQQueue queue = new AMQQueue(((AMQSession)session).getDefaultQueueExchangeName(), new AMQShortString("cluster_test_queue"));
topicProducer = session.createProducer(topic);
queueProducer = session.createProducer(queue);
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/flow/ChannelFlowTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/flow/ChannelFlowTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/flow/ChannelFlowTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/flow/ChannelFlowTest.java Fri Feb 16 15:11:41 2007
@@ -24,6 +24,8 @@
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.Message;
import javax.jms.MessageListener;
@@ -41,7 +43,7 @@
ChannelFlowTest(AMQConnection connection) throws Exception
{
- this(connection, new AMQQueue(randomize("ChannelFlowTest"), true));
+ this(connection, new AMQQueue(connection.getDefaultQueueExchangeName(), new AMQShortString(randomize("ChannelFlowTest")), true));
}
ChannelFlowTest(AMQConnection connection, AMQDestination destination) throws Exception
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java Fri Feb 16 15:11:41 2007
@@ -21,9 +21,12 @@
package org.apache.qpid.fragmentation;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.url.URLSyntaxException;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.client.AMQSession;
import org.apache.qpid.jms.MessageProducer;
import org.apache.qpid.jms.Session;
import org.apache.log4j.Logger;
@@ -49,7 +52,7 @@
private AMQConnection _connection;
- private Session _session;
+ private AMQSession _session;
private class CallbackHandler implements MessageListener
{
@@ -109,8 +112,8 @@
{
createConnection(host, port, clientID);
- _session = (Session) _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- AMQTopic destination = new AMQTopic("large");
+ _session = (AMQSession) _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ AMQTopic destination = new AMQTopic(_session.getDefaultTopicExchangeName(), new AMQShortString("large"));
MessageProducer producer = (MessageProducer) _session.createProducer(destination);
_connection.start();
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java Fri Feb 16 15:11:41 2007
@@ -22,7 +22,10 @@
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.client.AMQSession;
import org.apache.qpid.jms.Session;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
import org.apache.log4j.Logger;
import javax.jms.*;
@@ -76,11 +79,12 @@
InetAddress address = InetAddress.getLocalHost();
AMQConnection con = new AMQConnection(host, port, username, password,
address.getHostName(), virtualPath);
- final Session session = (Session) con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ final AMQSession session = (AMQSession) con.createSession(false, Session.AUTO_ACKNOWLEDGE);
final int expectedMessageCount = numExpectedMessages;
- MessageConsumer consumer = session.createConsumer(new AMQTopic("large"),
+ MessageConsumer consumer = session.createConsumer(new AMQTopic(session.getDefaultTopicExchangeName(),
+ new AMQShortString("large")),
100, true, false, null);
consumer.setMessageListener(new MessageListener()
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/latency/LatencyTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/latency/LatencyTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/latency/LatencyTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/latency/LatencyTest.java Fri Feb 16 15:11:41 2007
@@ -24,6 +24,8 @@
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.MessageProducer;
import javax.jms.Message;
@@ -51,7 +53,7 @@
LatencyTest(AMQConnection connection, int count, int delay, int length) throws Exception
{
- this(connection, new AMQQueue(randomize("LatencyTest"), true), count, delay, length);
+ this(connection, new AMQQueue(connection.getDefaultQueueExchangeName(), new AMQShortString(randomize("LatencyTest")), true), count, delay, length);
}
LatencyTest(AMQConnection connection, AMQDestination destination, int count, int delay, int length) throws Exception
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/multiconsumer/AMQTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/multiconsumer/AMQTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/multiconsumer/AMQTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/multiconsumer/AMQTest.java Fri Feb 16 15:11:41 2007
@@ -31,7 +31,6 @@
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
-import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
@@ -41,6 +40,9 @@
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.Session;
/**
* Test AMQ.
@@ -54,7 +56,7 @@
private static final String DUMMYCONTENT = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
private static final String HUGECONTENT;
- private Connection connect = null;
+ private AMQConnection connect = null;
private Session pubSession = null;
private Session subSession = null;
private Topic topic = null;
@@ -75,7 +77,7 @@
connect.setExceptionListener(this);
pubSession = connect.createSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
subSession = connect.createSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
- topic = new AMQTopic(SUBJECT);
+ topic = new AMQTopic(pubSession.getDefaultTopicExchangeName(), new AMQShortString(SUBJECT));
connect.start();
}
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestPublisher.java Fri Feb 16 15:11:41 2007
@@ -23,6 +23,8 @@
import org.apache.log4j.Logger;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.url.URLSyntaxException;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.jms.MessageProducer;
@@ -110,8 +112,8 @@
{
createConnection(host, port, clientID);
- _session = (Session) _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- AMQTopic destination = new AMQTopic(commandQueueName);
+ _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ AMQTopic destination = new AMQTopic(_session.getDefaultTopicExchangeName(), new AMQShortString(commandQueueName));
MessageProducer producer = (MessageProducer) _session.createProducer(destination);
_connection.start();
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestSubscriber.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestSubscriber.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestSubscriber.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/pubsub1/TestSubscriber.java Fri Feb 16 15:11:41 2007
@@ -24,6 +24,8 @@
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.jms.Session;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -87,17 +89,17 @@
InetAddress address = InetAddress.getLocalHost();
AMQConnection con1 = new AMQConnection(args[0], Integer.parseInt(args[1]), args[2], args[3],
address.getHostName(), args[4]);
- final org.apache.qpid.jms.Session session1 = (org.apache.qpid.jms.Session) con1.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ final Session session1 = con1.createSession(false, Session.AUTO_ACKNOWLEDGE);
AMQConnection con2 = new AMQConnection(args[0], Integer.parseInt(args[1]), args[2], args[3],
address.getHostName(), args[4]);
- final org.apache.qpid.jms.Session session2 = (org.apache.qpid.jms.Session) con2.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ final Session session2 = con2.createSession(false, Session.AUTO_ACKNOWLEDGE);
String selector = args[6];
final int expectedMessageCount = Integer.parseInt(args[5]);
_logger.info("Message selector is <" + selector + ">...");
- Topic t = new AMQTopic("cbr");
+ Topic t = new AMQTopic(session1.getDefaultTopicExchangeName(), new AMQShortString("cbr"));
MessageConsumer consumer1 = session1.createConsumer(t,
100, false, false, selector);
MessageConsumer consumer2 = session2.createConsumer(t,
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/topic/MessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/topic/MessageFactory.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/topic/MessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/topic/MessageFactory.java Fri Feb 16 15:11:41 2007
@@ -22,6 +22,8 @@
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
import javax.jms.*;
@@ -47,8 +49,8 @@
_session = session;
if(session instanceof AMQSession)
{
- _topic = new AMQTopic("topictest.messages");
- _control = new AMQTopic("topictest.control");
+ _topic = new AMQTopic(((AMQSession)session).getDefaultTopicExchangeName(),new AMQShortString("topictest.messages"));
+ _control = new AMQTopic(((AMQSession)session).getDefaultTopicExchangeName(),new AMQShortString("topictest.control"));
}
else
{
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Ping.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Ping.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Ping.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Ping.java Fri Feb 16 15:11:41 2007
@@ -22,6 +22,8 @@
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.client.AMQQueue;
import javax.jms.Connection;
@@ -35,7 +37,7 @@
Config config = new Config(argv);
Connection con = config.createConnection();
con.setClientID("ping");
- new Relay(new AMQQueue("ping"), new AMQQueue("pong"), con,
+ new Relay(new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString("ping")), new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString("pong")), con,
config.isEchoOn(),
config.getBatchSize(),
config.usePersistentMessages()).start();
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Pong.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Pong.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Pong.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Pong.java Fri Feb 16 15:11:41 2007
@@ -22,6 +22,8 @@
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.client.AMQQueue;
import javax.jms.Connection;
@@ -34,7 +36,7 @@
Config config = new Config(argv);
Connection con = config.createConnection();
con.setClientID("pong");
- new Relay(new AMQQueue("pong"), new AMQQueue("ping"), con,
+ new Relay(new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString("pong")), new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString("ping")), con,
config.isEchoOn(),
config.getBatchSize(),
config.usePersistentMessages()).start();
Modified: incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Start.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Start.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Start.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/old_test/java/org/apache/qpid/transacted/Start.java Fri Feb 16 15:11:41 2007
@@ -22,6 +22,8 @@
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.client.AMQQueue;
import javax.jms.Connection;
@@ -33,7 +35,7 @@
public static void main(String[] argv) throws Exception
{
Connection con = new Config(argv).createConnection();
- AMQQueue ping = new AMQQueue("ping");
+ AMQQueue ping = new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString("ping"));
Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
session.createProducer(ping).send(session.createTextMessage("start"));
session.close();
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java Fri Feb 16 15:11:41 2007
@@ -28,7 +28,7 @@
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
-import javax.jms.Session;
+
import javax.jms.TextMessage;
import junit.framework.TestCase;
@@ -40,6 +40,7 @@
import org.apache.qpid.client.transport.TransportConnection;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.Session;
public class RecoverTest extends TestCase
{
@@ -66,15 +67,15 @@
public void testRecoverResendsMsgs() throws Exception
{
- Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
+ AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue(new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
+ Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(),new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
//force synch to ensure the consumer has resulted in a bound queue
((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
- Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
+ AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
MessageProducer producer = producerSession.createProducer(queue);
@@ -123,15 +124,15 @@
public void testRecoverResendsMsgsAckOnEarlier() throws Exception
{
- Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
+ AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue(new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
+ Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("someQ"), new AMQShortString("someQ"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
//force synch to ensure the consumer has resulted in a bound queue
((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
- Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
+ AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
MessageProducer producer = producerSession.createProducer(queue);
@@ -187,15 +188,15 @@
public void testAcknowledgePerConsumer() throws Exception
{
- Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
+ AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- Queue queue = new AMQQueue(new AMQShortString("Q1"), new AMQShortString("Q1"), false, true);
- Queue queue2 = new AMQQueue(new AMQShortString("Q2"), new AMQShortString("Q2"), false, true);
+ Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q1"), new AMQShortString("Q1"), false, true);
+ Queue queue2 = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q2"), new AMQShortString("Q2"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
MessageConsumer consumer2 = consumerSession.createConsumer(queue2);
- Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
+ AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");
Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
MessageProducer producer = producerSession.createProducer(queue);
MessageProducer producer2 = producerSession.createProducer(queue2);
@@ -226,10 +227,10 @@
public void testRecoverInAutoAckListener() throws Exception
{
- Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
+ AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");
final Session consumerSession = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = new AMQQueue(new AMQShortString("Q3"), new AMQShortString("Q3"), false, true);
+ Queue queue = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q3"), new AMQShortString("Q3"), false, true);
MessageConsumer consumer = consumerSession.createConsumer(queue);
MessageProducer producer = consumerSession.createProducer(queue);
producer.send(consumerSession.createTextMessage("hello"));
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/BytesMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/BytesMessageTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/BytesMessageTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/BytesMessageTest.java Fri Feb 16 15:11:41 2007
@@ -68,7 +68,7 @@
void init(AMQConnection connection) throws Exception
{
- init(connection, new AMQQueue(randomize("BytesMessageTest"), true));
+ init(connection, new AMQQueue(connection, randomize("BytesMessageTest"), true));
}
void init(AMQConnection connection, AMQDestination destination) throws Exception
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java Fri Feb 16 15:11:41 2007
@@ -70,7 +70,7 @@
private void init(AMQConnection connection) throws Exception
{
- init(connection, new AMQQueue(randomize("FieldTableMessageTest"), true));
+ init(connection, new AMQQueue(connection,randomize("FieldTableMessageTest"), true));
}
private void init(AMQConnection connection, AMQDestination destination) throws Exception
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/LargeMessageTest.java Fri Feb 16 15:11:41 2007
@@ -68,7 +68,7 @@
private void init(AMQConnection connection) throws Exception
{
- Destination destination = new AMQQueue("LargeMessageTest", true);
+ Destination destination = new AMQQueue(connection,"LargeMessageTest", true);
init(connection, destination);
}
Modified: incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java?view=diff&rev=508649&r1=508648&r2=508649
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java Fri Feb 16 15:11:41 2007
@@ -83,7 +83,7 @@
private void init(AMQConnection connection) throws Exception
{
- Destination destination = new AMQQueue(randomize("MapMessageTest"), true);
+ Destination destination = new AMQQueue(connection,randomize("MapMessageTest"), true);
init(connection, destination);
}