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/01/24 16:41:57 UTC
svn commit: r499446 [1/4] - in /incubator/qpid/trunk/qpid/java: broker/etc/
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/configuration/
broker/src/main/java/org/apache/qpid/server/exchange/ broker/src/main/ja...
Author: rgreig
Date: Wed Jan 24 07:41:48 2007
New Revision: 499446
URL: http://svn.apache.org/viewvc?view=rev&rev=499446
Log:
QPID-50 : Patch supplied by Rob Godfrey - Virtual Host implementation
Added:
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/ManagedVirtualHost.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostRegistry.java
Modified:
incubator/qpid/trunk/qpid/java/broker/etc/config.xml
incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/Configurator.java
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/AbstractExchange.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.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/BasicQosHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.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/handler/QueueDeleteHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/TxRollbackHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/TxSelectHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/management/DefaultManagedObject.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPProtocolProvider.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/DefaultQueueRegistry.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueRegistry.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ConfigurationFileApplicationRegistry.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/state/StateAwareMethodListener.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/util/NullApplicationRegistry.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/old_test/java/org/apache/qpid/test/unit/client/connection/TestManyConnections.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/client/MessageListenerTest.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/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/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/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/message/ObjectMessageTest.java
incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.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/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/cluster/src/main/java/org/apache/qpid/server/cluster/ClusterBuilder.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusteredProtocolHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ClusteredProtocolSession.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/ServerHandlerRegistry.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ChainedClusterMethodHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ClusterMethodHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ExtendedHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/NullListener.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/PeerHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/QueueNameGenerator.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/RemoteCancelHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/RemoteConsumeHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ReplicatingConsumeHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/handler/WrappedListener.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/cluster/replay/ReplayStore.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/ClusteredQueue.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/PrivateQueue.java
incubator/qpid/trunk/qpid/java/cluster/src/main/java/org/apache/qpid/server/queue/RemoteQueueProxy.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/framing/BasicContentHeaderProperties.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/Constants.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/ServerRegistry.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/jmx/ClientListener.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/jmx/ClientNotificationListener.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/jmx/JMXServerRegistry.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/MBeanTypeTabControl.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/MBeanView.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/NavigationView.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/NotificationsTabControl.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/OperationTabControl.java
incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/views/ViewUtility.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/ServiceRequestingClient.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTest.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MockProtocolSession.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SkeletonMessageStore.java
incubator/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/util/TestApplicationRegistry.java
Modified: incubator/qpid/trunk/qpid/java/broker/etc/config.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/etc/config.xml?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/etc/config.xml (original)
+++ incubator/qpid/trunk/qpid/java/broker/etc/config.xml Wed Jan 24 07:41:48 2007
@@ -79,6 +79,37 @@
</mechanisms>
</sasl>
</security>
+ <virtualhosts>
+ <virtualhost>
+ <name>localhost</name>
+ <localhost>
+ <store>
+ <!-- <class>org.apache.qpid.server.store.berkeleydb.BDBMessageStore</class> -->
+ <class>org.apache.qpid.server.store.MemoryMessageStore</class>
+ <environment-path>localhost-store</environment-path>
+ </store>
+ </localhost>
+ </virtualhost>
+
+ <virtualhost>
+ <name>development</name>
+ <development>
+ <store>
+ <class>org.apache.qpid.server.store.MemoryMessageStore</class>
+ </store>
+ </development>
+ </virtualhost>
+
+ <virtualhost>
+ <name>test</name>
+ <test>
+ <store>
+ <class>org.apache.qpid.server.store.MemoryMessageStore</class>
+ </store>
+ </test>
+ </virtualhost>
+
+ </virtualhosts>
<heartbeat>
<delay>0</delay>
<timeoutFactor>2.0</timeoutFactor>
@@ -86,8 +117,6 @@
<queue>
<auto_register>true</auto_register>
</queue>
- <store>
- <class>org.apache.qpid.server.store.MemoryMessageStore</class>
- </store>
+
<virtualhosts>${conf}/virtualhosts.xml</virtualhosts>
</broker>
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=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml (original)
+++ incubator/qpid/trunk/qpid/java/broker/etc/virtualhosts.xml Wed Jan 24 07:41:48 2007
@@ -21,7 +21,17 @@
-->
<virtualhosts>
<virtualhost>
- <path>/development</path>
+ <path>localhost</path>
+ <bind>direct://amq.direct//queue</bind>
+ <bind>direct://amq.direct//ping</bind>
+ </virtualhost>
+ <virtualhost>
+ <path>development</path>
+ <bind>direct://amq.direct//queue</bind>
+ <bind>direct://amq.direct//ping</bind>
+ </virtualhost>
+ <virtualhost>
+ <path>test</path>
<bind>direct://amq.direct//queue</bind>
<bind>direct://amq.direct//ping</bind>
</virtualhost>
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java Wed Jan 24 07:41:48 2007
@@ -30,6 +30,7 @@
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
@@ -50,20 +51,28 @@
private final ExchangeFactory _exchangeFactory;
private final MessageStore _messageStore;
+ private final VirtualHost.VirtualHostMBean _virtualHostMBean;
+
@MBeanConstructor("Creates the Broker Manager MBean")
- public AMQBrokerManagerMBean() throws JMException
+ public AMQBrokerManagerMBean(VirtualHost.VirtualHostMBean virtualHostMBean) throws JMException
{
super(ManagedBroker.class, ManagedBroker.TYPE);
- IApplicationRegistry appRegistry = ApplicationRegistry.getInstance();
- _queueRegistry = appRegistry.getQueueRegistry();
- _exchangeRegistry = appRegistry.getExchangeRegistry();
- _exchangeFactory = ApplicationRegistry.getInstance().getExchangeFactory();
- _messageStore = ApplicationRegistry.getInstance().getMessageStore();
+
+ _virtualHostMBean = virtualHostMBean;
+ VirtualHost virtualHost = virtualHostMBean.getVirtualHost();
+
+
+
+
+ _queueRegistry = virtualHost.getQueueRegistry();
+ _exchangeRegistry = virtualHost.getExchangeRegistry();
+ _messageStore = virtualHost.getMessageStore();
+ _exchangeFactory = virtualHost.getExchangeFactory();
}
public String getObjectInstanceName()
{
- return this.getClass().getName();
+ return _virtualHostMBean.getVirtualHost().getName();
}
/**
@@ -144,7 +153,7 @@
try
{
- queue = new AMQQueue(new AMQShortString(queueName), durable, new AMQShortString(owner), autoDelete, _queueRegistry);
+ queue = new AMQQueue(new AMQShortString(queueName), durable, new AMQShortString(owner), autoDelete, getVirtualHost());
if (queue.isDurable() && !queue.isAutoDelete())
{
_messageStore.createQueue(queue);
@@ -157,6 +166,11 @@
}
}
+ private VirtualHost getVirtualHost()
+ {
+ return _virtualHostMBean.getVirtualHost();
+ }
+
/**
* Deletes the queue from queue registry and persistant storage.
*
@@ -183,11 +197,17 @@
}
}
- public ObjectName getObjectName() throws MalformedObjectNameException
+ public ManagedObject getParentObject()
{
- StringBuffer objectName = new StringBuffer(ManagedObject.DOMAIN);
- objectName.append(":type=").append(getType());
-
- return new ObjectName(objectName.toString());
+ return _virtualHostMBean;
}
+
+// public ObjectName getObjectName() throws MalformedObjectNameException
+// {
+// StringBuffer objectName = new StringBuffer(ManagedObject.DOMAIN);
+// objectName.append(".").append(getVirtualHost().getName());
+// objectName.append(":type=").append(getType());
+//
+// return new ObjectName(objectName.toString());
+// }
} // End of MBean class
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java Wed Jan 24 07:41:48 2007
@@ -62,7 +62,7 @@
* Main entry point for AMQPD.
*
*/
-public class Main implements ProtocolVersionList, Managable
+public class Main implements ProtocolVersionList
{
private static final Logger _logger = Logger.getLogger(Main.class);
@@ -70,7 +70,8 @@
private static final String DEFAULT_LOG_CONFIG_FILENAME = "log4j.xml";
- private AMQBrokerManagerMBean _mbean = null;
+
+ private static Main _instance;
protected static class InitException extends Exception
{
@@ -265,7 +266,6 @@
}
bind(port, connectorConfig);
- createAndRegisterBrokerMBean();
}
protected void setupVirtualHosts(String configFileParent, String configFilePath) throws ConfigurationException, AMQException, URLSyntaxException
@@ -368,7 +368,7 @@
public static void main(String[] args)
{
- new Main(args);
+ _instance = new Main(args);
}
private byte[] parseIP(String address) throws Exception
@@ -430,21 +430,4 @@
}
}
- private void createAndRegisterBrokerMBean() throws AMQException
- {
- try
- {
- _mbean = new AMQBrokerManagerMBean();
- _mbean.register();
- }
- catch (JMException ex)
- {
- throw new AMQException("Exception occured in creating AMQBrokerManager MBean");
- }
- }
-
- public ManagedObject getManagedObject()
- {
- return _mbean;
- }
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/Configurator.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/Configurator.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/Configurator.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/Configurator.java Wed Jan 24 07:41:48 2007
@@ -37,14 +37,15 @@
{
private static final Logger _logger = Logger.getLogger(Configurator.class);
+
/**
- * Configure a given instance using the application configuration. Note that superclasses are <b>not</b>
+ * Configure a given instance using the supplied configuration. Note that superclasses are <b>not</b>
* currently configured but this could easily be added if required.
* @param instance the instance to configure
+ * @param config the configuration to use to configure the object
*/
- public static void configure(Object instance)
+ public static void configure(Object instance, Configuration config)
{
- final Configuration config = ApplicationRegistry.getInstance().getConfiguration();
for (Field f : instance.getClass().getDeclaredFields())
{
@@ -54,6 +55,18 @@
setValueInField(f, instance, config, annotation);
}
}
+ }
+
+
+
+ /**
+ * Configure a given instance using the application configuration. Note that superclasses are <b>not</b>
+ * currently configured but this could easily be added if required.
+ * @param instance the instance to configure
+ */
+ public static void configure(Object instance)
+ {
+ configure(instance, ApplicationRegistry.getInstance().getConfiguration());
}
private static void setValueInField(Field f, Object instance, Configuration config, Configured annotation)
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=499446&r1=499445&r2=499446
==============================================================================
--- 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 Wed Jan 24 07:41:48 2007
@@ -28,6 +28,7 @@
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.log4j.Logger;
@@ -113,6 +114,7 @@
}
_logger.info("VirtualHost:'" + prop + "'");
+ String virtualHost = prop.toString();
prop = _config.getProperty(path + "." + XML_BIND);
if (prop instanceof Collection)
@@ -121,16 +123,16 @@
_logger.debug("Number of Bindings: " + bindings);
for (int dest = 0; dest < bindings; dest++)
{
- loadBinding(path, dest);
+ loadBinding(virtualHost, path, dest);
}
}
else
{
- loadBinding(path, -1);
+ loadBinding(virtualHost,path, -1);
}
}
- private void loadBinding(String rootpath, int index) throws AMQException, ConfigurationException, URLSyntaxException
+ private void loadBinding(String virtualHost, String rootpath, int index) throws AMQException, ConfigurationException, URLSyntaxException
{
String path = rootpath + "." + XML_BIND;
if (index != -1)
@@ -146,7 +148,7 @@
try
{
- bind(binding);
+ bind(virtualHost, binding);
}
catch (AMQException amqe)
{
@@ -155,7 +157,7 @@
}
}
- private void bind(AMQBindingURL binding) throws AMQException, ConfigurationException
+ private void bind(String virtualHostName, AMQBindingURL binding) throws AMQException, ConfigurationException
{
AMQShortString queueName = binding.getQueueName();
@@ -169,9 +171,10 @@
}
//Get references to Broker Registries
- QueueRegistry queueRegistry = ApplicationRegistry.getInstance().getQueueRegistry();
- MessageStore messageStore = ApplicationRegistry.getInstance().getMessageStore();
- ExchangeRegistry exchangeRegistry = ApplicationRegistry.getInstance().getExchangeRegistry();
+ VirtualHost virtualHost = ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHost(virtualHostName);
+ QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
+ MessageStore messageStore = virtualHost.getMessageStore();
+ ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry();
synchronized (queueRegistry)
{
@@ -184,7 +187,7 @@
queue = new AMQQueue(queueName,
Boolean.parseBoolean(binding.getOption(AMQBindingURL.OPTION_DURABLE)),
null /* These queues will have no owner */,
- false /* Therefore autodelete makes no sence */, queueRegistry);
+ false /* Therefore autodelete makes no sence */, virtualHost);
if (queue.isDurable())
{
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java Wed Jan 24 07:41:48 2007
@@ -25,6 +25,10 @@
import org.apache.qpid.server.management.AMQManagedObject;
import org.apache.qpid.server.management.Managable;
import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.management.ManagedObjectRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.registry.ApplicationRegistry;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
@@ -34,10 +38,14 @@
{
private AMQShortString _name;
+
+
protected boolean _durable;
protected String _exchangeType;
protected int _ticket;
+ private VirtualHost _virtualHost;
+
protected ExchangeMBean _exchangeMbean;
/**
@@ -57,6 +65,11 @@
super(ManagedExchange.class, ManagedExchange.TYPE);
}
+ public ManagedObject getParentObject()
+ {
+ return _virtualHost.getManagedObject();
+ }
+
public String getObjectInstanceName()
{
return _name.toString();
@@ -87,13 +100,17 @@
return _autoDelete;
}
- public ObjectName getObjectName() throws MalformedObjectNameException
+// public ObjectName getObjectName() throws MalformedObjectNameException
+// {
+// String objNameString = super.getObjectName().toString();
+// objNameString = objNameString + ",VirtualHost="+ _virtualHost.getName() +",ExchangeType=" + _exchangeType;
+// return new ObjectName(objNameString);
+// }
+
+ protected ManagedObjectRegistry getManagedObjectRegistry()
{
- String objNameString = super.getObjectName().toString();
- objNameString = objNameString + ",ExchangeType=" + _exchangeType;
- return new ObjectName(objNameString);
+ return ApplicationRegistry.getInstance().getManagedObjectRegistry();
}
-
} // End of MBean class
public AMQShortString getName()
@@ -108,8 +125,9 @@
*/
protected abstract ExchangeMBean createMBean() throws AMQException;
- public void initialise(AMQShortString name, boolean durable, int ticket, boolean autoDelete) throws AMQException
+ public void initialise(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete) throws AMQException
{
+ _virtualHost = host;
_name = name;
_durable = durable;
_autoDelete = autoDelete;
@@ -151,4 +169,13 @@
return _exchangeMbean;
}
+ public VirtualHost getVirtualHost()
+ {
+ return _virtualHost;
+ }
+
+ public QueueRegistry getQueueRegistry()
+ {
+ return getVirtualHost().getQueueRegistry();
+ }
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java Wed Jan 24 07:41:48 2007
@@ -24,6 +24,7 @@
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQChannelException;
import org.apache.qpid.AMQUnknownExchangeType;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
@@ -36,9 +37,11 @@
private static final Logger _logger = Logger.getLogger(DefaultExchangeFactory.class);
private Map<AMQShortString, Class<? extends Exchange>> _exchangeClassMap = new HashMap<AMQShortString, Class<? extends Exchange>>();
+ private final VirtualHost _host;
- public DefaultExchangeFactory()
+ public DefaultExchangeFactory(VirtualHost host)
{
+ _host = host;
_exchangeClassMap.put(ExchangeDefaults.DIRECT_EXCHANGE_CLASS, org.apache.qpid.server.exchange.DestNameExchange.class);
_exchangeClassMap.put(ExchangeDefaults.TOPIC_EXCHANGE_CLASS, org.apache.qpid.server.exchange.DestWildExchange.class);
_exchangeClassMap.put(ExchangeDefaults.HEADERS_EXCHANGE_CLASS, org.apache.qpid.server.exchange.HeadersExchange.class);
@@ -59,7 +62,7 @@
try
{
Exchange e = exchClass.newInstance();
- e.initialise(exchange, durable, ticket, autoDelete);
+ e.initialise(_host, exchange, durable, ticket, autoDelete);
return e;
}
catch (InstantiationException e)
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java Wed Jan 24 07:41:48 2007
@@ -109,7 +109,7 @@
public void createNewBinding(String queueName, String binding) throws JMException
{
- AMQQueue queue = ApplicationRegistry.getInstance().getQueueRegistry().getQueue(new AMQShortString(queueName));
+ AMQQueue queue = getQueueRegistry().getQueue(new AMQShortString(queueName));
if (queue == null)
{
throw new JMException("Queue \"" + queueName + "\" is not registered with the exchange.");
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java Wed Jan 24 07:41:48 2007
@@ -31,6 +31,7 @@
import org.apache.qpid.server.queue.AMQMessage;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import javax.management.JMException;
import javax.management.MBeanException;
@@ -110,7 +111,7 @@
public void createNewBinding(String queueName, String binding) throws JMException
{
- AMQQueue queue = ApplicationRegistry.getInstance().getQueueRegistry().getQueue(new AMQShortString(queueName));
+ AMQQueue queue = getQueueRegistry().getQueue(new AMQShortString(queueName));
if (queue == null)
throw new JMException("Queue \"" + queueName + "\" is not registered with the exchange.");
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java Wed Jan 24 07:41:48 2007
@@ -25,13 +25,14 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.virtualhost.VirtualHost;
public interface Exchange
{
AMQShortString getName();
AMQShortString getType();
- void initialise(AMQShortString name, boolean durable, int ticket, boolean autoDelete) throws AMQException;
+ void initialise(VirtualHost host, AMQShortString name, boolean durable, int ticket, boolean autoDelete) throws AMQException;
boolean isDurable();
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java Wed Jan 24 07:41:48 2007
@@ -44,7 +44,7 @@
private TabularType _bindinglistDataType = null;
private TabularDataSupport _bindingList = null;
- @MBeanConstructor("Creates an MBean for AMQ direct exchange")
+ @MBeanConstructor("Creates an MBean for AMQ fanout exchange")
public FanoutExchangeMBean() throws JMException
{
super();
@@ -86,7 +86,7 @@
public void createNewBinding(String queueName, String binding) throws JMException
{
- AMQQueue queue = ApplicationRegistry.getInstance().getQueueRegistry().getQueue(new AMQShortString(queueName));
+ AMQQueue queue = getQueueRegistry().getQueue(new AMQShortString(queueName));
if (queue == null)
{
throw new JMException("Queue \"" + queueName + "\" is not registered with the exchange.");
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java Wed Jan 24 07:41:48 2007
@@ -150,7 +150,7 @@
*/
public void createNewBinding(String queueName, String binding) throws JMException
{
- AMQQueue queue = ApplicationRegistry.getInstance().getQueueRegistry().getQueue(new AMQShortString(queueName));
+ AMQQueue queue = getQueueRegistry().getQueue(new AMQShortString(queueName));
if (queue == null)
{
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -29,6 +29,7 @@
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.log4j.Logger;
public class BasicAckMethodHandler implements StateAwareMethodListener<BasicAckBody>
@@ -46,10 +47,10 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<BasicAckBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicAckBody> evt) throws AMQException
{
+ AMQProtocolSession protocolSession = stateManager.getProtocolSession();
+
if (_log.isDebugEnabled())
{
_log.debug("Ack received on channel " + evt.getChannelId());
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -30,6 +30,7 @@
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.AMQException;
public class BasicCancelMethodHandler implements StateAwareMethodListener<BasicCancelBody>
@@ -45,10 +46,10 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<BasicCancelBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicCancelBody> evt) throws AMQException
{
+ AMQProtocolSession protocolSession = stateManager.getProtocolSession();
+
final AMQChannel channel = protocolSession.getChannel(evt.getChannelId());
final BasicCancelBody body = evt.getMethod();
channel.unsubscribeConsumer(protocolSession, body.consumerTag);
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -28,6 +28,8 @@
import org.apache.qpid.framing.*;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.ConsumerTagNotUniqueException;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.protocol.AMQProtocolSession;
@@ -53,14 +55,15 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession session,
- AMQMethodEvent<BasicConsumeBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicConsumeBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
+
BasicConsumeBody body = evt.getMethod();
final int channelId = evt.getChannelId();
AMQChannel channel = session.getChannel(channelId);
+ VirtualHost vHost = session.getVirtualHost();
if (channel == null)
{
_log.error("Channel " + channelId + " not found");
@@ -69,7 +72,7 @@
else
{
- AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : queueRegistry.getQueue(body.queue);
+ AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.queue);
if (queue == null)
{
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -10,6 +10,8 @@
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.ConsumerTagNotUniqueException;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQInvalidSelectorException;
import org.apache.qpid.protocol.AMQConstant;
@@ -31,12 +33,13 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession session,
- AMQMethodEvent<BasicGetBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicGetBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
+
BasicGetBody body = evt.getMethod();
final int channelId = evt.getChannelId();
+ VirtualHost vHost = session.getVirtualHost();
AMQChannel channel = session.getChannel(channelId);
if (channel == null)
@@ -46,7 +49,7 @@
}
else
{
- AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : queueRegistry.getQueue(body.queue);
+ AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.queue);
if (queue == null)
{
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=499446&r1=499445&r2=499446
==============================================================================
--- 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 Wed Jan 24 07:41:48 2007
@@ -28,6 +28,8 @@
import org.apache.qpid.framing.ChannelCloseBody;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.exchange.Exchange;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.protocol.AMQMethodEvent;
@@ -53,10 +55,10 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<BasicPublishBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicPublishBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
+
final BasicPublishBody body = evt.getMethod();
if (_log.isDebugEnabled())
@@ -70,7 +72,8 @@
body.exchange = ExchangeDefaults.DIRECT_EXCHANGE_NAME;
}
- Exchange e = exchangeRegistry.getExchange(body.exchange);
+ VirtualHost vHost = session.getVirtualHost();
+ Exchange e = vHost.getExchangeRegistry().getExchange(body.exchange);
// if the exchange does not exist we raise a channel exception
if (e == null)
{
@@ -82,8 +85,8 @@
// The partially populated BasicDeliver frame plus the received route body
// is stored in the channel. Once the final body frame has been received
// it is routed to the exchange.
- AMQChannel channel = protocolSession.getChannel(evt.getChannelId());
- channel.setPublishFrame(body, protocolSession);
+ AMQChannel channel = session.getChannel(evt.getChannelId());
+ channel.setPublishFrame(body, session);
}
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java Wed Jan 24 07:41:48 2007
@@ -28,6 +28,7 @@
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.AMQException;
@@ -40,9 +41,9 @@
return _instance;
}
- public void methodReceived(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges,
- AMQProtocolSession session, AMQMethodEvent<BasicQosBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicQosBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
session.getChannel(evt.getChannelId()).setPrefetchCount(evt.getMethod().prefetchCount);
session.getChannel(evt.getChannelId()).setPrefetchSize(evt.getMethod().prefetchSize);
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -27,6 +27,7 @@
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.framing.BasicRecoverBody;
import org.apache.qpid.AMQException;
import org.apache.log4j.Logger;
@@ -42,18 +43,18 @@
return _instance;
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<BasicRecoverBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicRecoverBody> evt) throws AMQException
{
- _logger.debug("Recover received on protocol session " + protocolSession + " and channel " + evt.getChannelId());
- AMQChannel channel = protocolSession.getChannel(evt.getChannelId());
+ AMQProtocolSession session = stateManager.getProtocolSession();
+
+ _logger.debug("Recover received on protocol session " + session + " and channel " + evt.getChannelId());
+ AMQChannel channel = session.getChannel(evt.getChannelId());
if (channel == null)
{
throw new AMQException("Unknown channel " + evt.getChannelId());
}
BasicRecoverBody body = evt.getMethod();
- channel.resend(protocolSession, body.requeue);
+ channel.resend(session, body.requeue);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java Wed Jan 24 07:41:48 2007
@@ -31,6 +31,7 @@
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
public class ChannelCloseHandler implements StateAwareMethodListener<ChannelCloseBody>
{
@@ -47,18 +48,17 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ChannelCloseBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelCloseBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
ChannelCloseBody body = evt.getMethod();
_logger.info("Received channel close for id " + evt.getChannelId() + " citing class " + body.classId +
" and method " + body.methodId);
- protocolSession.closeChannel(evt.getChannelId());
+ session.closeChannel(evt.getChannelId());
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
AMQFrame response = ChannelCloseOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0);
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java Wed Jan 24 07:41:48 2007
@@ -28,6 +28,7 @@
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.log4j.Logger;
public class ChannelCloseOkHandler implements StateAwareMethodListener<ChannelCloseOkBody>
@@ -45,9 +46,7 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ChannelCloseOkBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelCloseOkBody> evt) throws AMQException
{
_logger.info("Received channel-close-ok for channel-id " + evt.getChannelId());
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java Wed Jan 24 07:41:48 2007
@@ -48,13 +48,12 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ChannelFlowBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelFlowBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
ChannelFlowBody body = evt.getMethod();
- AMQChannel channel = protocolSession.getChannel(evt.getChannelId());
+ AMQChannel channel = session.getChannel(evt.getChannelId());
channel.setSuspended(!body.active);
_logger.debug("Channel.Flow for channel " + evt.getChannelId() + ", active=" + body.active);
@@ -64,6 +63,6 @@
AMQFrame response = ChannelFlowOkBody.createAMQFrame(evt.getChannelId(),
(byte)8, (byte)0, // AMQP version (major, minor)
body.active); // active
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java Wed Jan 24 07:41:48 2007
@@ -25,6 +25,7 @@
import org.apache.qpid.framing.ChannelOpenBody;
import org.apache.qpid.framing.ChannelOpenOkBody;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.exchange.ExchangeRegistry;
@@ -47,18 +48,18 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ChannelOpenBody> evt) throws AMQException
- {
- IApplicationRegistry registry = ApplicationRegistry.getInstance();
- final AMQChannel channel = new AMQChannel(evt.getChannelId(), registry.getMessageStore(),
- exchangeRegistry);
- protocolSession.addChannel(channel);
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelOpenBody> evt) throws AMQException
+ {
+ AMQProtocolSession session = stateManager.getProtocolSession();
+ VirtualHost virtualHost = session.getVirtualHost();
+
+ final AMQChannel channel = new AMQChannel(evt.getChannelId(), virtualHost.getMessageStore(),
+ virtualHost.getExchangeRegistry());
+ session.addChannel(channel);
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
AMQFrame response = ChannelOpenOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0);
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -47,16 +47,15 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ConnectionCloseBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionCloseBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
final ConnectionCloseBody body = evt.getMethod();
_logger.info("ConnectionClose received with reply code/reply text " + body.replyCode + "/" +
- body.replyText + " for " + protocolSession);
+ body.replyText + " for " + session);
try
{
- protocolSession.closeSession();
+ session.closeSession();
}
catch (Exception e)
{
@@ -66,6 +65,6 @@
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
final AMQFrame response = ConnectionCloseOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0);
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -46,17 +46,16 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ConnectionCloseOkBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionCloseOkBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
//todo should this not do more than just log the method?
_logger.info("Received Connection-close-ok");
try
{
stateManager.changeState(AMQState.CONNECTION_CLOSED);
- protocolSession.closeSession();
+ session.closeSession();
}
catch (Exception e)
{
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -27,11 +27,13 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.protocol.AMQMethodEvent;
+import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.state.AMQState;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.virtualhost.VirtualHost;
public class ConnectionOpenMethodHandler implements StateAwareMethodListener<ConnectionOpenBody>
{
@@ -51,28 +53,39 @@
return new AMQShortString(Long.toString(System.currentTimeMillis()));
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ConnectionOpenBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionOpenBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
ConnectionOpenBody body = evt.getMethod();
-
+ String virtualHostName = String.valueOf(body.virtualHost);
- //todo //FIXME The virtual host must be validated by the server for the connection to open-ok
- // See Spec (0.8.2). Section 3.1.2 Virtual Hosts
- if (protocolSession.getContextKey() == null)
+ VirtualHost virtualHost = stateManager.getVirtualHostRegistry().getVirtualHost(virtualHostName);
+
+ if(virtualHost == null)
{
- protocolSession.setContextKey(generateClientID());
+ throw body.getConnectionException(AMQConstant.NOT_FOUND.getCode(), "Unknown virtual host: " + virtualHostName);
}
+ else
+ {
+ session.setVirtualHost( virtualHost );
+
- // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
- // TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
- // Be aware of possible changes to parameter order as versions change.
- AMQFrame response = ConnectionOpenOkBody.createAMQFrame((short)0,
- (byte)8, (byte)0, // AMQP version (major, minor)
- body.virtualHost); // knownHosts
- stateManager.changeState(AMQState.CONNECTION_OPEN);
- protocolSession.writeFrame(response);
+
+ // See Spec (0.8.2). Section 3.1.2 Virtual Hosts
+ if (session.getContextKey() == null)
+ {
+ session.setContextKey(generateClientID());
+ }
+
+ // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
+ // TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
+ // Be aware of possible changes to parameter order as versions change.
+ AMQFrame response = ConnectionOpenOkBody.createAMQFrame((short)0,
+ (byte)8, (byte)0, // AMQP version (major, minor)
+ body.virtualHost);
+ stateManager.changeState(AMQState.CONNECTION_OPEN);
+ session.writeFrame(response);
+ }
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -54,14 +54,13 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ConnectionSecureOkBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionSecureOkBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
ConnectionSecureOkBody body = evt.getMethod();
AuthenticationManager authMgr = ApplicationRegistry.getInstance().getAuthenticationManager();
- SaslServer ss = protocolSession.getSaslServer();
+ SaslServer ss = session.getSaslServer();
if (ss == null)
{
throw new AMQException("No SASL context set up in session");
@@ -84,8 +83,8 @@
ConnectionCloseBody.getMethod((byte)8, (byte)0), // methodId
AMQConstant.NOT_ALLOWED.getCode(), // replyCode
AMQConstant.NOT_ALLOWED.getName()); // replyText
- protocolSession.writeFrame(close);
- disposeSaslServer(protocolSession);
+ session.writeFrame(close);
+ disposeSaslServer(session);
break;
case SUCCESS:
_logger.info("Connected as: " + ss.getAuthorizationID());
@@ -101,8 +100,8 @@
Integer.MAX_VALUE, // channelMax
ConnectionStartOkMethodHandler.getConfiguredFrameSize(), // frameMax
HeartbeatConfig.getInstance().getDelay()); // heartbeat
- protocolSession.writeFrame(tune);
- disposeSaslServer(protocolSession);
+ session.writeFrame(tune);
+ disposeSaslServer(session);
break;
case CONTINUE:
stateManager.changeState(AMQState.CONNECTION_NOT_AUTH);
@@ -112,7 +111,7 @@
AMQFrame challenge = ConnectionSecureBody.createAMQFrame(0,
(byte)8, (byte)0, // AMQP version (major, minor)
authResult.challenge); // challenge
- protocolSession.writeFrame(challenge);
+ session.writeFrame(challenge);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -60,10 +60,9 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ConnectionStartOkBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionStartOkBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
final ConnectionStartOkBody body = evt.getMethod();
_logger.info("SASL Mechanism selected: " + body.mechanism);
_logger.info("Locale selected: " + body.locale);
@@ -73,15 +72,15 @@
SaslServer ss = null;
try
{
- ss = authMgr.createSaslServer(String.valueOf(body.mechanism), protocolSession.getLocalFQDN());
- protocolSession.setSaslServer(ss);
+ ss = authMgr.createSaslServer(String.valueOf(body.mechanism), session.getLocalFQDN());
+ session.setSaslServer(ss);
AuthenticationResult authResult = authMgr.authenticate(ss, body.response);
//save clientProperties
- if (protocolSession.getClientProperties() == null)
+ if (session.getClientProperties() == null)
{
- protocolSession.setClientProperties(body.clientProperties);
+ session.setClientProperties(body.clientProperties);
}
switch (authResult.status)
@@ -100,7 +99,7 @@
Integer.MAX_VALUE, // channelMax
getConfiguredFrameSize(), // frameMax
HeartbeatConfig.getInstance().getDelay()); // heartbeat
- protocolSession.writeFrame(tune);
+ session.writeFrame(tune);
break;
case CONTINUE:
stateManager.changeState(AMQState.CONNECTION_NOT_AUTH);
@@ -110,12 +109,12 @@
AMQFrame challenge = ConnectionSecureBody.createAMQFrame(0,
(byte)8, (byte)0, // AMQP version (major, minor)
authResult.challenge); // challenge
- protocolSession.writeFrame(challenge);
+ session.writeFrame(challenge);
}
}
catch (SaslException e)
{
- disposeSaslServer(protocolSession);
+ disposeSaslServer(session);
throw new AMQException("SASL error: " + e, e);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java Wed Jan 24 07:41:48 2007
@@ -42,16 +42,15 @@
return _instance;
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ConnectionTuneOkBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionTuneOkBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
ConnectionTuneOkBody body = evt.getMethod();
if (_logger.isDebugEnabled())
{
_logger.debug(body);
}
stateManager.changeState(AMQState.CONNECTION_NOT_OPENED);
- protocolSession.initHeartbeats(body.heartbeat);
+ session.initHeartbeats(body.heartbeat);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java Wed Jan 24 07:41:48 2007
@@ -30,6 +30,7 @@
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.virtualhost.VirtualHost;
/**
* @author Apache Software Foundation
@@ -61,10 +62,12 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ExchangeBoundBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ExchangeBoundBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
+ VirtualHost virtualHost = session.getVirtualHost();
+ QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
+
// AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
byte major = (byte)8;
@@ -79,7 +82,7 @@
{
throw new AMQException("Exchange exchange must not be null");
}
- Exchange exchange = exchangeRegistry.getExchange(exchangeName);
+ Exchange exchange = virtualHost.getExchangeRegistry().getExchange(exchangeName);
AMQFrame response;
if (exchange == null)
{
@@ -112,6 +115,7 @@
}
else
{
+
AMQQueue queue = queueRegistry.getQueue(queueName);
if (queue == null)
{
@@ -194,6 +198,6 @@
" to exchange " + exchangeName)); // replyText
}
}
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java Wed Jan 24 07:41:48 2007
@@ -38,6 +38,7 @@
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
public class ExchangeDeclareHandler implements StateAwareMethodListener<ExchangeDeclareBody>
{
@@ -50,17 +51,19 @@
return _instance;
}
- private final ExchangeFactory exchangeFactory;
+
private ExchangeDeclareHandler()
{
- exchangeFactory = ApplicationRegistry.getInstance().getExchangeFactory();
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ExchangeDeclareBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ExchangeDeclareBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
+ VirtualHost virtualHost = session.getVirtualHost();
+ ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry();
+ ExchangeFactory exchangeFactory = virtualHost.getExchangeFactory();
+
final ExchangeDeclareBody body = evt.getMethod();
if (_logger.isDebugEnabled())
{
@@ -106,7 +109,7 @@
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
AMQFrame response = ExchangeDeclareOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0);
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
}
}
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java?view=diff&rev=499446&r1=499445&r2=499446
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java Wed Jan 24 07:41:48 2007
@@ -31,6 +31,7 @@
import org.apache.qpid.server.queue.QueueRegistry;
import org.apache.qpid.server.state.AMQStateManager;
import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.virtualhost.VirtualHost;
public class ExchangeDeleteHandler implements StateAwareMethodListener<ExchangeDeleteBody>
{
@@ -45,10 +46,12 @@
{
}
- public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
- ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
- AMQMethodEvent<ExchangeDeleteBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ExchangeDeleteBody> evt) throws AMQException
{
+ AMQProtocolSession session = stateManager.getProtocolSession();
+ VirtualHost virtualHost = session.getVirtualHost();
+ ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry();
+
ExchangeDeleteBody body = evt.getMethod();
try
{
@@ -57,7 +60,7 @@
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
AMQFrame response = ExchangeDeleteOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0);
- protocolSession.writeFrame(response);
+ session.writeFrame(response);
}
catch (ExchangeInUseException e)
{