Author: rgodfrey
Date: Fri Dec 11 15:50:19 2015
New Revision: 1719463
URL: http://svn.apache.org/viewvc?rev=1719463&view=rev
Log:
QPID-4943 : Remove old interfaces for queues, exchanges and virtual hosts
Removed:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/ExchangeImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBCacheSizeSetter.java
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHost.java
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHost.java
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHost.java
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8Test.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingFactory.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionPrincipal.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionVersionValidator.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/topic/TopicExchangeResult.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/filter/FilterSupport.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/logging/AbstractVirtualHostLogger.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/logging/VirtualHostFileLoggerImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/logging/VirtualHostSyslogLoggerImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/BindingLogSubject.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubject.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/QueueLogSubject.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Binding.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Consumer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Exchange.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHostNode.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionValidator.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/plugin/MessageConverter.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/plugin/SystemNodeCreator.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/LastValueQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/LastValueQueueImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/NotificationCheck.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/OrderedQueueEntryList.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueueImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueueImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueue.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/queue/StandardQueueImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/security/TrustStoreMessageSource.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/security/TrustStoreMessageSourceCreator.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AsynchronousMessageStoreRecoverer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ExchangeExistsException.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/MessageStoreRecoverer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/NonStandardVirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/ProvidedStoreVirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/QueueExistsException.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/SynchronousMessageStoreRecoverer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostPrincipal.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostPropertiesNode.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostUnavailableException.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostalias/VirtualHostNameAliasImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/AbstractStandardVirtualHostNode.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/AbstractVirtualHostNode.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/binding/BindingImplTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/configuration/startup/VirtualHostCreationTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/configuration/store/StoreConfigurationChangeListenerTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/exchange/DirectExchangeTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/exchange/FanoutExchangeTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/exchange/HeadersBindingTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/exchange/HeadersExchangeTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/VirtualHostLoggerTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/VirtualHostNameAndLevelLogInclusionRuleImplTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/actors/BaseConnectionActorTestCase.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/messages/ExchangeMessagesTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/AbstractTestLogSubject.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/BindingLogSubjectTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubjectTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubjectTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/QueueLogSubjectTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/VirtualHostTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/AbstractQueueTestBase.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/LastValueQueueListTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/LastValueQueueTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/NotificationCheckTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/PriorityQueueListTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryListTestBase.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/QueueMessageRecoveryTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleQueueEntryImplTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryListTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/SortedQueueEntryTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueEntryListTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/queue/StandardQueueTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/security/SecurityManagerTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/VirtualHostStoreUpgraderAndRecovererTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/util/BrokerTestHelper.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/virtualhost/AsynchronousMessageStoreRecovererTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/virtualhost/SynchronousMessageStoreRecovererTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostQueueCreationTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/virtualhostalias/VirtualHostAliasTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/virtualhostnode/AbstractStandardVirtualHostNodeTest.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10_to_Internal.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegateTest.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Test.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeueTest.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ExchangeDestination.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_to_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/MessageConverter_v1_0_to_Internal.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/QueueDestination.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ReceivingLink_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/TxnCoordinatorLink_1_0.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0Test.java
qpid/java/trunk/broker-plugins/amqp-msg-conv-0-10-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_10_v1_0/MessageConverter_1_0_to_v0_10.java
qpid/java/trunk/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_10_to_0_8.java
qpid/java/trunk/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_8_to_0_10.java
qpid/java/trunk/broker-plugins/amqp-msg-conv-0-8-to-1-0/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v1_0/MessageConverter_1_0_to_v0_8.java
qpid/java/trunk/broker-plugins/derby-store/src/main/java/org/apache/qpid/server/virtualhost/derby/DerbyVirtualHost.java
qpid/java/trunk/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/virtualhost/jdbc/JDBCVirtualHost.java
qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
qpid/java/trunk/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/QueueReportServlet.java
qpid/java/trunk/broker-plugins/management-http/src/test/java/org/apache/qpid/server/management/plugin/report/ReportRunnerTest.java
qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBCacheSizeSetter.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBCacheSizeSetter.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBCacheSizeSetter.java
(original)
+++
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBCacheSizeSetter.java
Fri Dec 11 15:50:19 2015
@@ -83,7 +83,7 @@ public class BDBCacheSizeSetter extends
}
}
- VirtualHost<?,?,?> virtualHost = virtualHostNode.getVirtualHost();
+ VirtualHost<?> virtualHost = virtualHostNode.getVirtualHost();
if (virtualHost instanceof BDBEnvironmentContainer &&
virtualHost.getState() == State.ACTIVE)
{
Long explicitJECacheSizeForVHost =
getExplicitJECacheSize(virtualHost);
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHost.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHost.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHost.java
(original)
+++
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHost.java
Fri Dec 11 15:50:19 2015
@@ -19,17 +19,14 @@
package org.apache.qpid.server.virtualhost.berkeleydb;
-import org.apache.qpid.server.exchange.ExchangeImpl;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.virtualhost.NonStandardVirtualHost;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
/**
Object that represents the VirtualHost whilst the VirtualHostNode is in the
replica role. The
real virtualhost will be elsewhere in the group.
*/
public interface BDBHAReplicaVirtualHost<X extends BDBHAReplicaVirtualHost<X>>
- extends VirtualHostImpl<X, AMQQueue<?>, ExchangeImpl<?>>,
- NonStandardVirtualHost<X,AMQQueue<?>,ExchangeImpl<?>>
+ extends VirtualHost<X>, NonStandardVirtualHost<X>
{
}
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
(original)
+++
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
Fri Dec 11 15:50:19 2015
@@ -31,7 +31,6 @@ import java.util.concurrent.ScheduledFut
import com.google.common.util.concurrent.ListenableFuture;
-import org.apache.qpid.server.exchange.ExchangeImpl;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
@@ -44,12 +43,12 @@ import org.apache.qpid.server.model.Exch
import org.apache.qpid.server.model.ManagedAttributeField;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.VirtualHostAlias;
import org.apache.qpid.server.model.VirtualHostNode;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.LinkRegistry;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.store.DurableConfigurationStore;
@@ -154,7 +153,7 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
- public ExchangeImpl createExchange(final Map<String, Object> attributes)
+ public Exchange<?> createExchange(final Map<String, Object> attributes)
{
throwUnsupportedForReplica();
return null;
@@ -167,13 +166,13 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
- public ExchangeImpl<?> getAttainedExchange(final String name)
+ public Exchange<?> getAttainedExchange(final String name)
{
return null;
}
@Override
- public AMQQueue<?> createQueue(final Map<String, Object> attributes)
+ public Queue<?> createQueue(final Map<String, Object> attributes)
{
throwUnsupportedForReplica();
return null;
@@ -312,7 +311,7 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
- public AMQQueue<?> getAttainedQueue(final String name)
+ public Queue<?> getAttainedQueue(final String name)
{
return null;
}
@@ -324,33 +323,33 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
- public AMQQueue<?> getAttainedQueue(final UUID id)
+ public Queue<?> getAttainedQueue(final UUID id)
{
return null;
}
@Override
- public Collection<AMQQueue<?>> getQueues()
+ public Collection<Queue<?>> getQueues()
{
return Collections.emptyList();
}
@Override
- public ListenableFuture<Integer> removeQueueAsync(final AMQQueue<?> queue)
+ public ListenableFuture<Integer> removeQueueAsync(final Queue<?> queue)
{
throwUnsupportedForReplica();
return null;
}
@Override
- public int removeQueue(final AMQQueue<?> queue)
+ public int removeQueue(final Queue<?> queue)
{
throwUnsupportedForReplica();
return 0;
}
@Override
- public Collection<ExchangeImpl<?>> getExchanges()
+ public Collection<Exchange<?>> getExchanges()
{
return Collections.emptyList();
}
@@ -403,35 +402,6 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
- public long getHouseKeepingTaskCount()
- {
- return 0;
- }
-
- @Override
- public long getHouseKeepingCompletedTaskCount()
- {
- return 0;
- }
-
- @Override
- public int getHouseKeepingPoolSize()
- {
- return 0;
- }
-
- @Override
- public void setHouseKeepingPoolSize(final int newSize)
- {
- }
-
- @Override
- public int getHouseKeepingActiveCount()
- {
- return 0;
- }
-
- @Override
public DtxRegistry getDtxRegistry()
{
return null;
@@ -451,12 +421,6 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
- public boolean getDefaultDeadLetterQueueEnabled()
- {
- return false;
- }
-
- @Override
public EventLogger getEventLogger()
{
return null;
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHost.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHost.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHost.java
(original)
+++
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHost.java
Fri Dec 11 15:50:19 2015
@@ -20,18 +20,16 @@
*/
package org.apache.qpid.server.virtualhost.berkeleydb;
-import org.apache.qpid.server.exchange.ExchangeImpl;
import org.apache.qpid.server.model.DerivedAttribute;
import org.apache.qpid.server.model.ManagedAttribute;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.store.SizeMonitoringSettings;
import org.apache.qpid.server.virtualhost.NonStandardVirtualHost;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
public interface BDBHAVirtualHost<X extends BDBHAVirtualHost<X>>
- extends VirtualHostImpl<X, AMQQueue<?>, ExchangeImpl<?>>,
+ extends VirtualHost<X>,
SizeMonitoringSettings,
- NonStandardVirtualHost<X,AMQQueue<?>,ExchangeImpl<?>>
+ NonStandardVirtualHost<X>
{
String REMOTE_TRANSACTION_SYNCHRONIZATION_POLICY =
"remoteTransactionSynchronizationPolicy";
String LOCAL_TRANSACTION_SYNCHRONIZATION_POLICY =
"localTransactionSynchronizationPolicy";
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHost.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHost.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHost.java
(original)
+++
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHost.java
Fri Dec 11 15:50:19 2015
@@ -20,15 +20,13 @@
package org.apache.qpid.server.virtualhost.berkeleydb;
-import org.apache.qpid.server.exchange.ExchangeImpl;
import org.apache.qpid.server.model.ManagedAttribute;
import org.apache.qpid.server.model.ManagedContextDefault;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.store.SizeMonitoringSettings;
import org.apache.qpid.server.store.berkeleydb.BDBEnvironmentContainer;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
-public interface BDBVirtualHost<X extends BDBVirtualHost<X>> extends
VirtualHostImpl<X, AMQQueue<?>, ExchangeImpl<?>>,
org.apache.qpid.server.store.FileBasedSettings, SizeMonitoringSettings,
BDBEnvironmentContainer
+public interface BDBVirtualHost<X extends BDBVirtualHost<X>> extends
VirtualHost<X>, org.apache.qpid.server.store.FileBasedSettings,
SizeMonitoringSettings, BDBEnvironmentContainer
{
String STORE_PATH = "storePath";
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
(original)
+++
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
Fri Dec 11 15:50:19 2015
@@ -617,7 +617,7 @@ public class BDBHAVirtualHostNodeImpl ex
upgraderAndRecoverer.perform(getConfigurationStore());
getEventLogger().message(getConfigurationStoreLogSubject(),
ConfigStoreMessages.RECOVERY_COMPLETE());
- VirtualHost<?,?,?> host = getVirtualHost();
+ VirtualHost<?> host = getVirtualHost();
if (host == null)
{
@@ -637,7 +637,7 @@ public class BDBHAVirtualHostNodeImpl ex
host = getVirtualHost();
if(host != null)
{
- final VirtualHost<?,?,?> recoveredHost = host;
+ final VirtualHost<?> recoveredHost = host;
Subject.doAs(SecurityManager.getSubjectWithAddedSystemRights(), new
PrivilegedAction<Object>()
{
@Override
@@ -667,7 +667,7 @@ public class BDBHAVirtualHostNodeImpl ex
LOGGER.debug("Recovered virtualhost with name : " +
getGroupName());
}
- final VirtualHost<?,?,?> recoveredHost = host;
+ final VirtualHost<?> recoveredHost = host;
Subject.doAs(SecurityManager.getSubjectWithAddedSystemRights(), new
PrivilegedAction<Object>()
{
@Override
@@ -729,7 +729,7 @@ public class BDBHAVirtualHostNodeImpl ex
protected ListenableFuture<Void> closeVirtualHostIfExist()
{
- final VirtualHost<?,?,?> virtualHost = getVirtualHost();
+ final VirtualHost<?> virtualHost = getVirtualHost();
if (virtualHost!= null)
{
return doAfter(virtualHost.closeAsync(), new Runnable()
Modified:
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
(original)
+++
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
Fri Dec 11 15:50:19 2015
@@ -135,7 +135,7 @@ public class BDBHAVirtualHostNodeTest ex
assertEquals("SYNC,NO_SYNC,SIMPLE_MAJORITY",
environmentFacade.getMessageStoreDurability().toString());
_helper.awaitForVirtualhost(node, 30000);
- VirtualHost<?, ?, ?> virtualHost = node.getVirtualHost();
+ VirtualHost<?> virtualHost = node.getVirtualHost();
assertNotNull("Virtual host child was not added", virtualHost);
assertEquals("Unexpected virtual host name", groupName,
virtualHost.getName());
assertEquals("Unexpected virtual host store",
bdbConfigurationStore.getMessageStore(), virtualHost.getMessageStore());
Modified:
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8Test.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8Test.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8Test.java
(original)
+++
qpid/java/trunk/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom7To8Test.java
Fri Dec 11 15:50:19 2015
@@ -51,9 +51,9 @@ public class UpgradeFrom7To8Test extends
private static final String CONFIGURED_OBJECT_HIERARCHY_DB_NAME =
"CONFIGURED_OBJECT_HIERARCHY";
@Override
- public VirtualHost<?,?,?> getVirtualHost()
+ public VirtualHost<?> getVirtualHost()
{
- VirtualHost<?,?,?> virtualHost = mock(VirtualHost.class);
+ VirtualHost<?> virtualHost = mock(VirtualHost.class);
when(virtualHost.getName()).thenReturn("test");
return virtualHost;
}
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingFactory.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingFactory.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingFactory.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingFactory.java
Fri Dec 11 15:50:19 2015
@@ -22,13 +22,11 @@ package org.apache.qpid.server.binding;
import java.util.Map;
-import org.apache.qpid.server.exchange.ExchangeImpl;
import org.apache.qpid.server.model.AbstractConfiguredObjectTypeFactory;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.plugin.PluggableService;
-import org.apache.qpid.server.queue.AMQQueue;
@PluggableService
public class BindingFactory extends
AbstractConfiguredObjectTypeFactory<BindingImpl>
@@ -41,8 +39,8 @@ public class BindingFactory extends Abst
@Override
protected BindingImpl createInstance(final Map<String, Object> attributes,
final ConfiguredObject<?>... parents)
{
- ExchangeImpl<?> exchange = (ExchangeImpl<?>) getParent(Exchange.class,
parents);
- AMQQueue<?> queue = (AMQQueue<?>) getParent(Queue.class, parents);
+ Exchange<?> exchange = (Exchange<?>) getParent(Exchange.class,
parents);
+ Queue<?> queue = (Queue<?>) getParent(Queue.class, parents);
BindingImpl binding = new BindingImpl(attributes, queue, exchange);
exchange.addBinding(binding);
return binding;
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingImpl.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingImpl.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingImpl.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/binding/BindingImpl.java
Fri Dec 11 15:50:19 2015
@@ -32,7 +32,6 @@ import com.google.common.util.concurrent
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.configuration.updater.Task;
import org.apache.qpid.server.exchange.AbstractExchange;
-import org.apache.qpid.server.exchange.ExchangeImpl;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.FilterSupport;
import org.apache.qpid.server.logging.EventLogger;
@@ -41,11 +40,11 @@ import org.apache.qpid.server.logging.su
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.ManagedAttributeField;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.StateTransition;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.security.SecurityManager;
public class BindingImpl
@@ -53,14 +52,14 @@ public class BindingImpl
implements org.apache.qpid.server.model.Binding<BindingImpl>
{
private final String _bindingKey;
- private final AMQQueue _queue;
- private final ExchangeImpl _exchange;
+ private final Queue<?> _queue;
+ private final Exchange<?> _exchange;
@ManagedAttributeField
private Map<String, Object> _arguments;
private final AtomicLong _matches = new AtomicLong();
private BindingLogSubject _logSubject;
- public BindingImpl(Map<String, Object> attributes, AMQQueue queue,
ExchangeImpl exchange)
+ public BindingImpl(Map<String, Object> attributes, Queue<?> queue,
Exchange<?> exchange)
{
super(parentsMap(queue,exchange),stripEmptyArguments(enhanceWithDurable(attributes,
queue, exchange)));
_bindingKey = getName();
@@ -103,8 +102,8 @@ public class BindingImpl
}
private static Map<String, Object> enhanceWithDurable(Map<String, Object>
attributes,
- final AMQQueue queue,
- final ExchangeImpl
exchange)
+ final Queue<?> queue,
+ final Exchange<?>
exchange)
{
if(!attributes.containsKey(DURABLE))
{
@@ -114,12 +113,13 @@ public class BindingImpl
return attributes;
}
+ @Override
public String getBindingKey()
{
return _bindingKey;
}
- public AMQQueue getAMQQueue()
+ public Queue<?> getAMQQueue()
{
return _queue;
}
@@ -131,7 +131,7 @@ public class BindingImpl
}
@Override
- public ExchangeImpl<?> getExchange()
+ public Exchange<?> getExchange()
{
return _exchange;
}
@@ -259,7 +259,7 @@ public class BindingImpl
{
authoriseCreate(this);
- AMQQueue queue = getAMQQueue();
+ Queue<?> queue = getAMQQueue();
Map<String, Object> arguments = getArguments();
if (arguments!=null && !arguments.isEmpty() &&
FilterSupport.argumentsContainFilter(arguments))
{
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionPrincipal.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionPrincipal.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionPrincipal.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionPrincipal.java
Fri Dec 11 15:50:19 2015
@@ -52,7 +52,7 @@ public class ConnectionPrincipal impleme
return _connection;
}
- public VirtualHost<?,?,?> getVirtualHost()
+ public VirtualHost<?> getVirtualHost()
{
return _connection.getVirtualHost();
}
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionVersionValidator.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionVersionValidator.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionVersionValidator.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/connection/ConnectionVersionValidator.java
Fri Dec 11 15:50:19 2015
@@ -72,7 +72,7 @@ public class ConnectionVersionValidator
@Override
public boolean validateConnectionCreation(final AMQPConnection<?>
connection,
- final VirtualHost<?, ?, ?>
virtualHost)
+ final VirtualHost<?> virtualHost)
{
String connectionVersion = connection.getClientVersion();
if (connectionVersion == null)
@@ -97,7 +97,7 @@ public class ConnectionVersionValidator
return valid;
}
- private boolean connectionMatches(VirtualHost<?, ?, ?> virtualHost, String
listName, final String connectionVersion)
+ private boolean connectionMatches(VirtualHost<?> virtualHost, String
listName, final String connectionVersion)
{
final List<String> versionRegexList = getContextValueList(virtualHost,
listName);
if (versionRegexList != null)
@@ -123,7 +123,7 @@ public class ConnectionVersionValidator
return false;
}
- private List<String> getContextValueList(final VirtualHost<?,?,?>
virtualHost, final String variableName)
+ private List<String> getContextValueList(final VirtualHost<?> virtualHost,
final String variableName)
{
if (virtualHost.getContextKeys(false).contains(variableName))
{
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
Fri Dec 11 15:50:19 2015
@@ -33,7 +33,6 @@ import java.util.concurrent.ConcurrentHa
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.util.concurrent.FutureCallback;
@@ -64,7 +63,7 @@ import org.apache.qpid.server.model.Publ
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.StateTransition;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -73,12 +72,11 @@ import org.apache.qpid.server.util.Actio
import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
import org.apache.qpid.server.virtualhost.RequiredExchangeException;
import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException;
public abstract class AbstractExchange<T extends AbstractExchange<T>>
extends AbstractConfiguredObject<T>
- implements ExchangeImpl<T>
+ implements Exchange<T>
{
private static final Logger _logger =
LoggerFactory.getLogger(AbstractExchange.class);
private final AtomicBoolean _closed = new AtomicBoolean();
@@ -86,9 +84,7 @@ public abstract class AbstractExchange<T
@ManagedAttributeField(beforeSet = "preSetAlternateExchange", afterSet =
"postSetAlternateExchange" )
private Exchange<?> _alternateExchange;
- private VirtualHostImpl _virtualHost;
-
- private final List<Action<ExchangeImpl>> _closeTaskList = new
CopyOnWriteArrayList<>();
+ private VirtualHost<?> _virtualHost;
/**
* Whether the exchange is automatically deleted once all queues have
detached from it
@@ -99,8 +95,7 @@ public abstract class AbstractExchange<T
private LogSubject _logSubject;
private Map<ExchangeReferrer,Object> _referrers = new
ConcurrentHashMap<ExchangeReferrer,Object>();
- private final CopyOnWriteArrayList<BindingImpl> _bindings = new
CopyOnWriteArrayList<BindingImpl>();
- private final AtomicInteger _bindingCountHigh = new AtomicInteger();
+ private final CopyOnWriteArrayList<Binding<?>> _bindings = new
CopyOnWriteArrayList<>();
private final AtomicLong _receivedMessageCount = new AtomicLong();
private final AtomicLong _receivedMessageSize = new AtomicLong();
private final AtomicLong _routedMessageCount = new AtomicLong();
@@ -108,11 +103,9 @@ public abstract class AbstractExchange<T
private final AtomicLong _droppedMessageCount = new AtomicLong();
private final AtomicLong _droppedMessageSize = new AtomicLong();
- private final CopyOnWriteArrayList<ExchangeImpl.BindingListener>
_listeners = new CopyOnWriteArrayList<ExchangeImpl.BindingListener>();
-
- private final ConcurrentMap<BindingIdentifier, BindingImpl> _bindingsMap =
new ConcurrentHashMap<BindingIdentifier, BindingImpl>();
+ private final ConcurrentMap<BindingIdentifier, Binding<?>> _bindingsMap =
new ConcurrentHashMap<>();
- public AbstractExchange(Map<String, Object> attributes, VirtualHostImpl
vhost)
+ public AbstractExchange(Map<String, Object> attributes, VirtualHost<?>
vhost)
{
super(parentsMap(vhost), attributes);
Set<String> providedAttributeNames = new
HashSet<>(attributes.keySet());
@@ -142,12 +135,8 @@ public abstract class AbstractExchange<T
private boolean isReservedExchangeName(String name)
{
- if (name == null || ExchangeDefaults.DEFAULT_EXCHANGE_NAME.equals(name)
- || name.startsWith("amq.") || name.startsWith("qpid."))
- {
- return true;
- }
- return false;
+ return name == null ||
ExchangeDefaults.DEFAULT_EXCHANGE_NAME.equals(name)
+ || name.startsWith("amq.") || name.startsWith("qpid.");
}
@@ -187,8 +176,8 @@ public abstract class AbstractExchange<T
{
List<ListenableFuture<Void>> removeBindingFutures = new
ArrayList<>(_bindings.size());
- List<BindingImpl> bindings = new ArrayList<>(_bindings);
- for(BindingImpl binding : bindings)
+ List<Binding<?>> bindings = new ArrayList<>(_bindings);
+ for(Binding<?> binding : bindings)
{
removeBindingFutures.add(binding.deleteAsync());
}
@@ -201,16 +190,11 @@ public abstract class AbstractExchange<T
{
if (_alternateExchange != null)
{
- ((ExchangeImpl)
_alternateExchange).removeReference(AbstractExchange.this);
+
_alternateExchange.removeReference(AbstractExchange.this);
}
getEventLogger().message(_logSubject,
ExchangeMessages.DELETED());
- for (Action<ExchangeImpl> task : _closeTaskList)
- {
- task.performAction(AbstractExchange.this);
- }
- _closeTaskList.clear();
deleted();
}
});
@@ -228,16 +212,16 @@ public abstract class AbstractExchange<T
return getClass().getSimpleName() + "[" + getName() +"]";
}
- public VirtualHostImpl getVirtualHost()
+ public VirtualHost<?> getVirtualHost()
{
return _virtualHost;
}
- public final boolean isBound(String bindingKey, Map<String,Object>
arguments, AMQQueue queue)
+ public final boolean isBound(String bindingKey, Map<String,Object>
arguments, Queue<?> queue)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
- if(bindingKey.equals(b.getBindingKey()) && queue ==
b.getAMQQueue())
+ if(bindingKey.equals(b.getBindingKey()) && queue == b.getQueue())
{
return (b.getArguments() == null || b.getArguments().isEmpty())
? (arguments == null || arguments.isEmpty())
@@ -247,11 +231,11 @@ public abstract class AbstractExchange<T
return false;
}
- public final boolean isBound(String bindingKey, AMQQueue queue)
+ public final boolean isBound(String bindingKey, Queue<?> queue)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
- if(bindingKey.equals(b.getBindingKey()) && queue ==
b.getAMQQueue())
+ if(bindingKey.equals(b.getBindingKey()) && queue == b.getQueue())
{
return true;
}
@@ -261,7 +245,7 @@ public abstract class AbstractExchange<T
public final boolean isBound(String bindingKey)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
if(bindingKey.equals(b.getBindingKey()))
{
@@ -271,11 +255,11 @@ public abstract class AbstractExchange<T
return false;
}
- public final boolean isBound(AMQQueue queue)
+ public final boolean isBound(Queue<?> queue)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
- if(queue == b.getAMQQueue())
+ if(queue == b.getQueue())
{
return true;
}
@@ -284,11 +268,11 @@ public abstract class AbstractExchange<T
}
@Override
- public final boolean isBound(Map<String, Object> arguments, AMQQueue queue)
+ public final boolean isBound(Map<String, Object> arguments, Queue<?> queue)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
- if(queue == b.getAMQQueue() &&
+ if(queue == b.getQueue() &&
((b.getArguments() == null || b.getArguments().isEmpty())
? (arguments == null || arguments.isEmpty())
: b.getArguments().equals(arguments)))
@@ -302,7 +286,7 @@ public abstract class AbstractExchange<T
public final boolean isBound(Map<String, Object> arguments)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
if(((b.getArguments() == null || b.getArguments().isEmpty())
? (arguments == null || arguments.isEmpty())
@@ -318,7 +302,7 @@ public abstract class AbstractExchange<T
@Override
public final boolean isBound(String bindingKey, Map<String, Object>
arguments)
{
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
if(b.getBindingKey().equals(bindingKey) &&
((b.getArguments() == null || b.getArguments().isEmpty())
@@ -345,15 +329,16 @@ public abstract class AbstractExchange<T
{
if (_alternateExchange != null)
{
- ((ExchangeImpl) _alternateExchange).removeReference(this);
+ _alternateExchange.removeReference(this);
}
}
+ @SuppressWarnings("unused")
private void postSetAlternateExchange()
{
if(_alternateExchange != null)
{
- ((ExchangeImpl)_alternateExchange).addReference(this);
+ _alternateExchange.addReference(this);
}
}
@@ -372,55 +357,26 @@ public abstract class AbstractExchange<T
return !_referrers.isEmpty();
}
- public void addCloseTask(final Action<ExchangeImpl> task)
- {
- _closeTaskList.add(task);
- }
-
- public void removeCloseTask(final Action<ExchangeImpl> task)
- {
- _closeTaskList.remove(task);
- }
-
public final void doAddBinding(final BindingImpl binding)
{
_bindings.add(binding);
- int bindingCountSize = _bindings.size();
- int maxBindingsSize;
- while((maxBindingsSize = _bindingCountHigh.get()) < bindingCountSize)
- {
- _bindingCountHigh.compareAndSet(maxBindingsSize, bindingCountSize);
- }
- for(BindingListener listener : _listeners)
- {
- listener.bindingAdded(this, binding);
- }
onBind(binding);
}
- public long getBindingCountHigh()
- {
- return _bindingCountHigh.get();
- }
-
- public final void doRemoveBinding(final BindingImpl binding)
+ public final void doRemoveBinding(final Binding<?> binding)
{
onUnbind(binding);
- for(BindingListener listener : _listeners)
- {
- listener.bindingRemoved(this, binding);
- }
_bindings.remove(binding);
}
- public final Collection<BindingImpl> getBindings()
+ public final Collection<Binding<?>> getBindings()
{
return Collections.unmodifiableList(_bindings);
}
- protected abstract void onBind(final BindingImpl binding);
+ protected abstract void onBind(final Binding<?> binding);
- protected abstract void onUnbind(final BindingImpl binding);
+ protected abstract void onUnbind(final Binding<?> binding);
public Map<String, Object> getArguments()
{
@@ -451,7 +407,7 @@ public abstract class AbstractExchange<T
if(!deletedQueues)
{
deletedQueues = true;
- queues = new ArrayList<BaseQueue>(allQueues);
+ queues = new ArrayList<>(allQueues);
}
_logger.debug("Exchange: {} - attempt to enqueue message onto
deleted queue {}", getName(), q.getName());
@@ -584,18 +540,8 @@ public abstract class AbstractExchange<T
return _droppedMessageSize.get();
}
- public void addBindingListener(final BindingListener listener)
- {
- _listeners.add(listener);
- }
-
- public void removeBindingListener(final BindingListener listener)
- {
- _listeners.remove(listener);
- }
-
@Override
- public boolean addBinding(final String bindingKey, final AMQQueue queue,
final Map<String, Object> arguments)
+ public boolean addBinding(final String bindingKey, final Queue<?> queue,
final Map<String, Object> arguments)
{
return doSync(doOnConfigThread(new Task<ListenableFuture<Boolean>,
RuntimeException>()
{
@@ -629,7 +575,7 @@ public abstract class AbstractExchange<T
@Override
public boolean replaceBinding(final String bindingKey,
- final AMQQueue queue,
+ final Queue<?> queue,
final Map<String, Object> arguments)
{
return doSync(doOnConfigThread(new Task<ListenableFuture<Boolean>,
RuntimeException>()
@@ -638,7 +584,7 @@ public abstract class AbstractExchange<T
public ListenableFuture<Boolean> execute()
{
- final BindingImpl existingBinding = getBinding(bindingKey,
queue);
+ final Binding<?> existingBinding = getBinding(bindingKey,
queue);
return makeBindingAsync(existingBinding == null ? null :
existingBinding.getId(),
bindingKey,
queue,
@@ -667,10 +613,10 @@ public abstract class AbstractExchange<T
}
@Override
- public ListenableFuture<Void> removeBindingAsync(final BindingImpl binding)
+ public ListenableFuture<Void> removeBindingAsync(final Binding<?> binding)
{
String bindingKey = binding.getBindingKey();
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
assert queue != null;
@@ -682,7 +628,7 @@ public abstract class AbstractExchange<T
// Check access
authoriseDelete(binding);
- BindingImpl b = _bindingsMap.remove(new
BindingIdentifier(bindingKey,queue));
+ Binding<?> b = _bindingsMap.remove(new
BindingIdentifier(bindingKey,queue));
if (b != null)
{
@@ -727,7 +673,7 @@ public abstract class AbstractExchange<T
&& getBindingCount() == 0;
}
- public BindingImpl getBinding(String bindingKey, AMQQueue queue)
+ public Binding<?> getBinding(String bindingKey, Queue<?> queue)
{
assert queue != null;
@@ -741,7 +687,7 @@ public abstract class AbstractExchange<T
private ListenableFuture<Boolean> makeBindingAsync(UUID id,
String bindingKey,
- AMQQueue queue,
+ Queue<?> queue,
Map<String, Object> arguments,
boolean force)
{
@@ -759,7 +705,7 @@ public abstract class AbstractExchange<T
id = UUID.randomUUID();
}
- BindingImpl existingMapping;
+ Binding<?> existingMapping;
synchronized(this)
{
BindingIdentifier bindingIdentifier = new
BindingIdentifier(bindingKey, queue);
@@ -805,7 +751,7 @@ public abstract class AbstractExchange<T
else if(force)
{
Map<String,Object> oldArguments =
existingMapping.getArguments();
- existingMapping.setArguments(arguments);
+ ((BindingImpl)existingMapping).setArguments(arguments);
onBindingUpdated(existingMapping, oldArguments);
return Futures.immediateFuture(true);
}
@@ -817,17 +763,17 @@ public abstract class AbstractExchange<T
}
@Override
- public void addBinding(final BindingImpl b)
+ public void addBinding(final Binding<?> b)
{
- BindingIdentifier identifier = new BindingIdentifier(b.getName(),
b.getAMQQueue());
+ BindingIdentifier identifier = new BindingIdentifier(b.getName(),
b.getQueue());
_bindingsMap.put(identifier, b);
- b.getAMQQueue().addBinding(b);
+ b.getQueue().addBinding(b);
childAdded(b);
}
- protected abstract void onBindingUpdated(final BindingImpl binding,
+ protected abstract void onBindingUpdated(final Binding<?> binding,
final Map<String, Object>
oldArguments);
@@ -886,12 +832,18 @@ public abstract class AbstractExchange<T
}
}
+ private interface BindingListener<X extends AbstractExchange<X>>
+ {
+ void bindingAdded(AbstractExchange<X> exchange, Binding<?> binding);
+ void bindingRemoved(AbstractExchange<X> exchange, Binding<?> binding);
+ }
+
private static final class BindingIdentifier
{
private final String _bindingKey;
- private final AMQQueue _destination;
+ private final Queue<?> _destination;
- private BindingIdentifier(final String bindingKey, final AMQQueue
destination)
+ private BindingIdentifier(final String bindingKey, final Queue<?>
destination)
{
_bindingKey = bindingKey;
_destination = destination;
@@ -902,7 +854,7 @@ public abstract class AbstractExchange<T
return _bindingKey;
}
- public AMQQueue getDestination()
+ public Queue<?> getDestination()
{
return _destination;
}
@@ -950,9 +902,9 @@ public abstract class AbstractExchange<T
// Used by the protocol layers
@Override
- public boolean deleteBinding(final String bindingKey, final AMQQueue queue)
+ public boolean deleteBinding(final String bindingKey, final Queue<?> queue)
{
- final BindingImpl binding = getBinding(bindingKey, queue);
+ final Binding<?> binding = getBinding(bindingKey, queue);
if(binding == null)
{
return false;
@@ -966,7 +918,7 @@ public abstract class AbstractExchange<T
}
@Override
- public boolean hasBinding(final String bindingKey, final AMQQueue queue)
+ public boolean hasBinding(final String bindingKey, final Queue<?> queue)
{
return getBinding(bindingKey,queue) != null;
}
@@ -977,8 +929,8 @@ public abstract class AbstractExchange<T
final
Map<String, Object> bindingArguments,
final
Map<String, Object> attributes)
{
- addBinding(bindingKey, (AMQQueue) queue, bindingArguments);
- final BindingImpl binding = getBinding(bindingKey, (AMQQueue) queue);
+ addBinding(bindingKey, (Queue<?>) queue, bindingArguments);
+ final Binding<?> binding = getBinding(bindingKey, (Queue<?>) queue);
return binding;
}
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultDestination.java
Fri Dec 11 15:50:19 2015
@@ -24,19 +24,20 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
public class DefaultDestination implements MessageDestination
{
- private VirtualHostImpl _virtualHost;
+ private VirtualHost<?> _virtualHost;
- public DefaultDestination(VirtualHostImpl virtualHost)
+ public DefaultDestination(VirtualHost<?> virtualHost)
{
_virtualHost = virtualHost;
}
@@ -58,14 +59,14 @@ public class DefaultDestination implemen
{
routingAddress = "";
}
- final AMQQueue q = _virtualHost.getAttainedQueue(routingAddress);
+ final Queue<?> q = _virtualHost.getAttainedQueue(routingAddress);
if(q == null)
{
routingAddress = _virtualHost.getLocalAddress(routingAddress);
if(routingAddress.contains("/") && !routingAddress.startsWith("/"))
{
String[] parts = routingAddress.split("/",2);
- ExchangeImpl exchange =
_virtualHost.getAttainedExchange(parts[0]);
+ Exchange<?> exchange =
_virtualHost.getAttainedExchange(parts[0]);
if(exchange != null)
{
return exchange.send(message, parts[1],
instanceProperties, txn, postEnqueueAction);
@@ -73,7 +74,7 @@ public class DefaultDestination implemen
}
else if(!routingAddress.contains("/"))
{
- ExchangeImpl exchange =
_virtualHost.getAttainedExchange(routingAddress);
+ Exchange<?> exchange =
_virtualHost.getAttainedExchange(routingAddress);
if(exchange != null)
{
return exchange.send(message, "", instanceProperties, txn,
postEnqueueAction);
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java
Fri Dec 11 15:50:19 2015
@@ -35,18 +35,18 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.server.binding.BindingImpl;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.FilterSupport;
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.BaseQueue;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@ManagedObject( category = false, type =
ExchangeDefaults.DIRECT_EXCHANGE_CLASS )
public class DirectExchange extends AbstractExchange<DirectExchange>
@@ -56,23 +56,23 @@ public class DirectExchange extends Abst
private static final class BindingSet
{
- private CopyOnWriteArraySet<BindingImpl> _bindings = new
CopyOnWriteArraySet<BindingImpl>();
+ private CopyOnWriteArraySet<Binding<?>> _bindings = new
CopyOnWriteArraySet<>();
private List<BaseQueue> _unfilteredQueues = new ArrayList<BaseQueue>();
private Map<BaseQueue, FilterManager> _filteredQueues = new
HashMap<>();
- public synchronized void addBinding(BindingImpl binding)
+ public synchronized void addBinding(Binding<?> binding)
{
_bindings.add(binding);
recalculateQueues();
}
- public synchronized void removeBinding(BindingImpl binding)
+ public synchronized void removeBinding(Binding<?> binding)
{
_bindings.remove(binding);
recalculateQueues();
}
- public synchronized void updateBinding(BindingImpl binding)
+ public synchronized void updateBinding(Binding<?> binding)
{
recalculateQueues();
}
@@ -82,19 +82,19 @@ public class DirectExchange extends Abst
List<BaseQueue> queues = new
ArrayList<BaseQueue>(_bindings.size());
Map<BaseQueue, FilterManager> filteredQueues = new HashMap<>();
- for(BindingImpl b : _bindings)
+ for(Binding<?> b : _bindings)
{
if(FilterSupport.argumentsContainFilter(b.getArguments()))
{
try
{
- FilterManager filter =
FilterSupport.createMessageFilter(b.getArguments(), b.getAMQQueue());
- filteredQueues.put(b.getAMQQueue(),filter);
+ FilterManager filter =
FilterSupport.createMessageFilter(b.getArguments(), b.getQueue());
+ filteredQueues.put(b.getQueue(),filter);
}
catch (AMQInvalidArgumentException e)
{
- _logger.warn("Binding ignored: cannot parse filter on
binding of queue '"+b.getAMQQueue().getName()
+ _logger.warn("Binding ignored: cannot parse filter on
binding of queue '"+b.getQueue().getName()
+ "' to exchange '" +
b.getExchange().getName()
+ "' with arguments: " +
b.getArguments(), e);
}
@@ -103,9 +103,9 @@ public class DirectExchange extends Abst
else
{
- if(!queues.contains(b.getAMQQueue()))
+ if(!queues.contains(b.getQueue()))
{
- queues.add(b.getAMQQueue());
+ queues.add(b.getQueue());
}
}
}
@@ -119,7 +119,7 @@ public class DirectExchange extends Abst
return _unfilteredQueues;
}
- public CopyOnWriteArraySet<BindingImpl> getBindings()
+ public CopyOnWriteArraySet<Binding<?>> getBindings()
{
return _bindings;
}
@@ -139,7 +139,7 @@ public class DirectExchange extends Abst
new ConcurrentHashMap<String, BindingSet>();
@ManagedObjectFactoryConstructor
- public DirectExchange(final Map<String, Object> attributes, final
VirtualHostImpl vhost)
+ public DirectExchange(final Map<String, Object> attributes, final
VirtualHost<?> vhost)
{
super(attributes, vhost);
}
@@ -189,10 +189,10 @@ public class DirectExchange extends Abst
}
@Override
- protected void onBindingUpdated(final BindingImpl binding, final
Map<String, Object> oldArguments)
+ protected void onBindingUpdated(final Binding<?> binding, final
Map<String, Object> oldArguments)
{
String bindingKey = binding.getBindingKey();
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
assert queue != null;
assert bindingKey != null;
@@ -201,10 +201,10 @@ public class DirectExchange extends Abst
bindings.updateBinding(binding);
}
- protected void onBind(final BindingImpl binding)
+ protected void onBind(final Binding<?> binding)
{
String bindingKey = binding.getBindingKey();
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
assert queue != null;
assert bindingKey != null;
@@ -225,7 +225,7 @@ public class DirectExchange extends Abst
}
- protected void onUnbind(final BindingImpl binding)
+ protected void onUnbind(final Binding<?> binding)
{
assert binding != null;
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
Fri Dec 11 15:50:19 2015
@@ -31,18 +31,18 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.server.binding.BindingImpl;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.FilterSupport;
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.BaseQueue;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@ManagedObject( category = false, type =
ExchangeDefaults.FANOUT_EXCHANGE_CLASS )
public class FanoutExchange extends AbstractExchange<FanoutExchange>
@@ -54,19 +54,19 @@ public class FanoutExchange extends Abst
/**
* Maps from queue name to queue instances
*/
- private final Map<AMQQueue,Integer> _queues = new
HashMap<AMQQueue,Integer>();
- private final CopyOnWriteArrayList<AMQQueue> _unfilteredQueues = new
CopyOnWriteArrayList<AMQQueue>();
- private final CopyOnWriteArrayList<AMQQueue> _filteredQueues = new
CopyOnWriteArrayList<AMQQueue>();
+ private final Map<Queue<?>,Integer> _queues = new HashMap<>();
+ private final CopyOnWriteArrayList<Queue<?>> _unfilteredQueues = new
CopyOnWriteArrayList<>();
+ private final CopyOnWriteArrayList<Queue<?>> _filteredQueues = new
CopyOnWriteArrayList<>();
- private final AtomicReference<Map<AMQQueue,Map<BindingImpl,
FilterManager>>> _filteredBindings =
+ private final AtomicReference<Map<Queue<?>,Map<Binding<?>,
FilterManager>>> _filteredBindings =
new AtomicReference<>();
{
- Map<AMQQueue,Map<BindingImpl, FilterManager>> emptyMap =
Collections.emptyMap();
+ Map<Queue<?>,Map<Binding<?>, FilterManager>> emptyMap =
Collections.emptyMap();
_filteredBindings.set(emptyMap);
}
@ManagedObjectFactoryConstructor
- public FanoutExchange(final Map<String, Object> attributes, final
VirtualHostImpl vhost)
+ public FanoutExchange(final Map<String, Object> attributes, final
VirtualHost<?> vhost)
{
super(attributes, vhost);
}
@@ -77,7 +77,7 @@ public class FanoutExchange extends Abst
final InstanceProperties
instanceProperties)
{
- for(BindingImpl b : getBindings())
+ for(Binding<?> b : getBindings())
{
b.incrementMatches();
}
@@ -85,12 +85,12 @@ public class FanoutExchange extends Abst
final ArrayList<BaseQueue> result = new
ArrayList<BaseQueue>(_unfilteredQueues);
- final Map<AMQQueue, Map<BindingImpl, FilterManager>> filteredBindings
= _filteredBindings.get();
+ final Map<Queue<?>, Map<Binding<?>, FilterManager>> filteredBindings =
_filteredBindings.get();
if(!_filteredQueues.isEmpty())
{
- for(AMQQueue q : _filteredQueues)
+ for(Queue<?> q : _filteredQueues)
{
- final Map<BindingImpl, FilterManager> bindingMessageFilterMap
= filteredBindings.get(q);
+ final Map<Binding<?>, FilterManager> bindingMessageFilterMap =
filteredBindings.get(q);
if(!(bindingMessageFilterMap == null || result.contains(q)))
{
for(FilterManager filter :
bindingMessageFilterMap.values())
@@ -114,9 +114,9 @@ public class FanoutExchange extends Abst
}
@Override
- protected synchronized void onBindingUpdated(final BindingImpl binding,
final Map<String, Object> oldArguments)
+ protected synchronized void onBindingUpdated(final Binding<?> binding,
final Map<String, Object> oldArguments)
{
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
if (binding.getArguments() == null || binding.getArguments().isEmpty()
|| !FilterSupport.argumentsContainFilter(
binding.getArguments()))
@@ -140,16 +140,16 @@ public class FanoutExchange extends Abst
}
else
{
- HashMap<AMQQueue,Map<BindingImpl, FilterManager>> filteredBindings
=
+ HashMap<Queue<?>,Map<Binding<?>, FilterManager>> filteredBindings =
new HashMap<>(_filteredBindings.get());
- Map<BindingImpl,FilterManager> bindingsForQueue;
+ Map<Binding<?>,FilterManager> bindingsForQueue;
final FilterManager messageFilter;
try
{
- messageFilter =
FilterSupport.createMessageFilter(binding.getArguments(),
binding.getAMQQueue());
+ messageFilter =
FilterSupport.createMessageFilter(binding.getArguments(), binding.getQueue());
}
catch (AMQInvalidArgumentException e)
{
@@ -160,11 +160,11 @@ public class FanoutExchange extends Abst
if (oldArguments != null && !oldArguments.isEmpty() &&
FilterSupport.argumentsContainFilter(oldArguments))
{
- bindingsForQueue = new
HashMap<>(filteredBindings.remove(binding.getAMQQueue()));
+ bindingsForQueue = new
HashMap<>(filteredBindings.remove(binding.getQueue()));
}
else // previously unfiltered
{
- bindingsForQueue = new HashMap<BindingImpl,FilterManager>();
+ bindingsForQueue = new HashMap<>();
Integer oldValue = _queues.remove(queue);
if (ONE.equals(oldValue))
@@ -180,7 +180,7 @@ public class FanoutExchange extends Abst
}
bindingsForQueue.put(binding, messageFilter);
- filteredBindings.put(binding.getAMQQueue(),bindingsForQueue);
+ filteredBindings.put(binding.getQueue(),bindingsForQueue);
_filteredBindings.set(filteredBindings);
@@ -188,9 +188,10 @@ public class FanoutExchange extends Abst
}
- protected synchronized void onBind(final BindingImpl binding)
+ @Override
+ protected synchronized void onBind(final Binding<?> binding)
{
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
assert queue != null;
if(binding.getArguments() == null || binding.getArguments().isEmpty()
|| !FilterSupport.argumentsContainFilter(binding.getArguments()))
{
@@ -214,12 +215,12 @@ public class FanoutExchange extends Abst
try
{
- HashMap<AMQQueue,Map<BindingImpl, FilterManager>>
filteredBindings =
+ HashMap<Queue<?>,Map<Binding<?>, FilterManager>>
filteredBindings =
new HashMap<>(_filteredBindings.get());
- Map<BindingImpl, FilterManager> bindingsForQueue =
filteredBindings.remove(binding.getAMQQueue());
+ Map<Binding<?>, FilterManager> bindingsForQueue =
filteredBindings.remove(binding.getQueue());
final FilterManager messageFilter =
-
FilterSupport.createMessageFilter(binding.getArguments(),
binding.getAMQQueue());
+
FilterSupport.createMessageFilter(binding.getArguments(), binding.getQueue());
if(bindingsForQueue != null)
{
@@ -228,14 +229,14 @@ public class FanoutExchange extends Abst
}
else
{
- bindingsForQueue = Collections.singletonMap(binding,
messageFilter);
+ bindingsForQueue = Collections.<Binding<?>,
FilterManager>singletonMap(binding, messageFilter);
if(!_unfilteredQueues.contains(queue))
{
_filteredQueues.add(queue);
}
}
- filteredBindings.put(binding.getAMQQueue(), bindingsForQueue);
+ filteredBindings.put(binding.getQueue(), bindingsForQueue);
_filteredBindings.set(filteredBindings);
@@ -253,9 +254,10 @@ public class FanoutExchange extends Abst
}
}
- protected synchronized void onUnbind(final BindingImpl binding)
+ @Override
+ protected synchronized void onUnbind(final Binding<?> binding)
{
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
if(binding.getArguments() == null || binding.getArguments().isEmpty()
|| !FilterSupport.argumentsContainFilter(binding.getArguments()))
{
Integer oldValue = _queues.remove(queue);
@@ -275,15 +277,15 @@ public class FanoutExchange extends Abst
}
else // we are removing a binding with filters
{
- HashMap<AMQQueue,Map<BindingImpl, FilterManager>> filteredBindings
=
+ HashMap<Queue<?>,Map<Binding<?>, FilterManager>> filteredBindings =
new HashMap<>(_filteredBindings.get());
- Map<BindingImpl,FilterManager> bindingsForQueue =
filteredBindings.remove(binding.getAMQQueue());
+ Map<Binding<?>,FilterManager> bindingsForQueue =
filteredBindings.remove(binding.getQueue());
if(bindingsForQueue.size()>1)
{
bindingsForQueue = new HashMap<>(bindingsForQueue);
bindingsForQueue.remove(binding);
- filteredBindings.put(binding.getAMQQueue(),bindingsForQueue);
+ filteredBindings.put(binding.getQueue(),bindingsForQueue);
}
else
{
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersBinding.java
Fri Dec 11 15:50:19 2015
@@ -36,6 +36,7 @@ import org.apache.qpid.server.filter.Fil
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.model.Binding;
/**
* Defines binding and matching based on a set of headers.
@@ -45,7 +46,7 @@ class HeadersBinding
private static final Logger _logger =
LoggerFactory.getLogger(HeadersBinding.class);
private final Map<String,Object> _mappings;
- private final BindingImpl _binding;
+ private final Binding<?> _binding;
private final Set<String> required = new HashSet<String>();
private final Map<String,Object> matches = new HashMap<String,Object>();
private boolean matchAny;
@@ -59,7 +60,7 @@ class HeadersBinding
*
* @param binding the binding to create a header binding using
*/
- public HeadersBinding(BindingImpl binding)
+ public HeadersBinding(Binding<?> binding)
{
_binding = binding;
if(_binding !=null)
@@ -80,11 +81,11 @@ class HeadersBinding
{
try
{
- _filter =
FilterSupport.createMessageFilter(_mappings,_binding.getAMQQueue());
+ _filter =
FilterSupport.createMessageFilter(_mappings,_binding.getQueue());
}
catch (AMQInvalidArgumentException e)
{
- _logger.warn("Invalid filter in binding queue
'"+_binding.getAMQQueue().getName()
+ _logger.warn("Invalid filter in binding queue
'"+_binding.getQueue().getName()
+"' to exchange
'"+_binding.getExchange().getName()
+"' with arguments: " + _binding.getArguments());
_filter = new FilterManager();
@@ -111,7 +112,7 @@ class HeadersBinding
}
}
- public BindingImpl getBinding()
+ public Binding<?> getBinding()
{
return _binding;
}
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
URL:
http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java?rev=1719463&r1=1719462&r2=1719463&view=diff
==============================================================================
---
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
(original)
+++
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
Fri Dec 11 15:50:19 2015
@@ -33,15 +33,15 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.server.binding.BindingImpl;
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
-import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.queue.BaseQueue;
-import org.apache.qpid.server.virtualhost.VirtualHostImpl;
/**
* An exchange that binds queues based on a set of required headers and header
values
@@ -76,14 +76,14 @@ public class HeadersExchange extends Abs
private static final Logger _logger =
LoggerFactory.getLogger(HeadersExchange.class);
- private final ConcurrentMap<String, CopyOnWriteArraySet<BindingImpl>>
_bindingsByKey =
- new ConcurrentHashMap<String,
CopyOnWriteArraySet<BindingImpl>>();
+ private final ConcurrentMap<String, CopyOnWriteArraySet<Binding<?>>>
_bindingsByKey =
+ new ConcurrentHashMap<>();
private final CopyOnWriteArrayList<HeadersBinding> _bindingHeaderMatchers =
new CopyOnWriteArrayList<HeadersBinding>();
@ManagedObjectFactoryConstructor
- public HeadersExchange(final Map<String, Object> attributes, final
VirtualHostImpl vhost)
+ public HeadersExchange(final Map<String, Object> attributes, final
VirtualHost<?> vhost)
{
super(attributes, vhost);
}
@@ -101,36 +101,37 @@ public class HeadersExchange extends Abs
{
if
(hb.matches(Filterable.Factory.newInstance(payload,instanceProperties)))
{
- BindingImpl b = hb.getBinding();
+ Binding<?> b = hb.getBinding();
b.incrementMatches();
if (_logger.isDebugEnabled())
{
_logger.debug("Exchange " + getName() + ": delivering
message with headers " +
- payload.getMessageHeader() + " to " +
b.getAMQQueue().getName());
+ payload.getMessageHeader() + " to " +
b.getQueue().getName());
}
- queues.add(b.getAMQQueue());
+ queues.add(b.getQueue());
}
}
- return new ArrayList<BaseQueue>(queues);
+ return new ArrayList<>(queues);
}
- protected void onBind(final BindingImpl binding)
+ @Override
+ protected void onBind(final Binding<?> binding)
{
String bindingKey = binding.getBindingKey();
- AMQQueue queue = binding.getAMQQueue();
+ Queue<?> queue = binding.getQueue();
assert queue != null;
assert bindingKey != null;
- CopyOnWriteArraySet<BindingImpl> bindings =
_bindingsByKey.get(bindingKey);
+ CopyOnWriteArraySet<Binding<?>> bindings =
_bindingsByKey.get(bindingKey);
if(bindings == null)
{
- bindings = new CopyOnWriteArraySet<BindingImpl>();
- CopyOnWriteArraySet<BindingImpl> newBindings;
+ bindings = new CopyOnWriteArraySet<>();
+ CopyOnWriteArraySet<Binding<?>> newBindings;
if((newBindings = _bindingsByKey.putIfAbsent(bindingKey,
bindings)) != null)
{
bindings = newBindings;
@@ -149,7 +150,7 @@ public class HeadersExchange extends Abs
}
@Override
- protected void onBindingUpdated(final BindingImpl binding, final
Map<String, Object> oldArguments)
+ protected void onBindingUpdated(final Binding<?> binding, final
Map<String, Object> oldArguments)
{
HeadersBinding headersBinding = new HeadersBinding(binding);
ListIterator<HeadersBinding> iter =
_bindingHeaderMatchers.listIterator();
@@ -163,11 +164,11 @@ public class HeadersExchange extends Abs
}
- protected void onUnbind(final BindingImpl binding)
+ protected void onUnbind(final Binding<?> binding)
{
assert binding != null;
- CopyOnWriteArraySet<BindingImpl> bindings =
_bindingsByKey.get(binding.getBindingKey());
+ CopyOnWriteArraySet<Binding<?>> bindings =
_bindingsByKey.get(binding.getBindingKey());
if(bindings != null)
{
bindings.remove(binding);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
