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 2008/04/21 16:04:00 UTC
svn commit: r650148 [1/10] - in
/incubator/qpid/branches/broker-queue-refactor/java: broker/ broker/etc/
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/ack/
broker/src/main/java/org/apache/qpid/server/configura...
Author: rgodfrey
Date: Mon Apr 21 07:03:35 2008
New Revision: 650148
URL: http://svn.apache.org/viewvc?rev=650148&view=rev
Log:
Initial checkpoint of queue refactoring work
Added:
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DirectExchange.java (contents, props changed)
- copied, changed from r629806, incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/TopicExchange.java (contents, props changed)
- copied, changed from r629806, incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKey.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderKeyDictionary.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeaderMatcherResult.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersMatcherDFAState.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/headers/HeadersParser.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherDFAState.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicMatcherResult.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicParser.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWord.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/topic/TopicWordDictionary.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/AbstractFlowCreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueFactory.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryAgent.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/IncomingMessage.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/store/DerbyMessageStore.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription.java (contents, props changed)
- copied, changed from r641230, incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactory.java (contents, props changed)
- copied, changed from r641230, incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactoryImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java (contents, props changed)
- copied, changed from r641230, incubator/qpid/branches/M2.1/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java
Removed:
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessage.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DestNameExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DestWildExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/Subscription.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionFactory.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeightedSubscriptionManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/CleanupMessageOperation.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/DeliveryManagerTest.java
Modified:
incubator/qpid/branches/broker-queue-refactor/java/broker/etc/config.xml
incubator/qpid/branches/broker-queue-refactor/java/broker/etc/virtualhosts.xml
incubator/qpid/branches/broker-queue-refactor/java/broker/python-test.xml
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/Exchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/FanoutExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/HeadersExchange.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/Index.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/MessageRouter.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/NoRouteException.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/QueueUnbindHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessageHandle.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/ExchangeBindings.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/InMemoryMessageHandle.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/MessageHandleFactory.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/QueueEntry.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionManager.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/WeakReferenceMessageHandle.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/store/MessageStore.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/NonTransactionalContext.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/txn/TransactionalContext.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/MessageStoreTool.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Copy.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Dump.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/List.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Move.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Purge.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Select.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/tools/messagestore/commands/Show.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/protocol/TestMinaProtocolSession.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/basic/SessionStartTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Client.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/CombinedTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Service.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/ServiceCreator.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/filter/codec/OurCumulativeProtocolDecoder.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/transport/socket/nio/MultiThreadSocketConnector.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/qpid/common/ClientProperties.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/qpid/framing/AMQTypedValue.java
incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
incubator/qpid/branches/broker-queue-refactor/java/management/eclipse-plugin/src/test/java/org/apache/qpid/management/ui/ManagementConsoleTest.java
incubator/qpid/branches/broker-queue-refactor/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/diagnostic/DiagnosticExchange.java
incubator/qpid/branches/broker-queue-refactor/java/plugins/src/main/java/org/apache/qpid/extras/exchanges/example/TestExchange.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/ConcurrencyTestDisabled.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/SubscriptionManagerTest.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/store/SkeletonMessageStore.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
incubator/qpid/branches/broker-queue-refactor/java/systests/src/main/java/org/apache/qpid/test/client/DupsOkTest.java
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/etc/config.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/etc/config.xml?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/etc/config.xml (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/etc/config.xml Mon Apr 21 07:03:35 2008
@@ -17,21 +17,13 @@
- KIND, either express or implied. See the License for the
- specific language governing permissions and limitations
- under the License.
- -
+ - AMQ
-->
<broker>
<prefix>${QPID_HOME}</prefix>
<work>${QPID_WORK}</work>
<conf>${prefix}/etc</conf>
<connector>
- <!-- Uncomment out this block and edit the keystorePath and keystorePassword
- to enable SSL support
- <ssl>
- <enabled>true</enabled>
- <sslOnly>true</sslOnly>
- <keystorePath>/path/to/keystore.ks</keystorePath>
- <keystorePassword>keystorepass</keystorePassword>
- </ssl>-->
<qpidnio>false</qpidnio>
<protectio>
<enabled>false</enabled>
@@ -87,6 +79,7 @@
<name>localhost</name>
<localhost>
<store>
+ <!-- <class>org.apache.qpid.server.store.MemoryMessageStore</class> -->
<class>org.apache.qpid.server.store.MemoryMessageStore</class>
</store>
@@ -110,6 +103,7 @@
<name>test</name>
<test>
<store>
+ <!-- <class>org.apache.qpid.server.store.MemoryMessageStore</class> -->
<class>org.apache.qpid.server.store.MemoryMessageStore</class>
</store>
</test>
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/etc/virtualhosts.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/etc/virtualhosts.xml?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/etc/virtualhosts.xml (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/etc/virtualhosts.xml Mon Apr 21 07:03:35 2008
@@ -26,14 +26,16 @@
<localhost>
<exchanges>
<exchange>
- <type>direct</type>
- <name>test.direct</name>
- <durable>true</durable>
+ <name>testdirect</name>
+ <testdirect>
+ <durable>true</durable>
+ </testdirect>
</exchange>
<exchange>
- <type>topic</type>
- <name>test.topic</name>
+ <name>testtopic</name>
</exchange>
+
+
</exchanges>
<queues>
<exchange>amq.direct</exchange>
@@ -69,6 +71,17 @@
<virtualhost>
<name>development</name>
<development>
+ <exchanges>
+ <exchange>
+ <name>testdirect</name>
+ <testdirect>
+ <durable>true</durable>
+ </testdirect>
+ </exchange>
+ <exchange>
+ <name>testtopic</name>
+ </exchange>
+ </exchanges>
<queues>
<minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
<maximumMessageCount>5000</maximumMessageCount>
@@ -85,6 +98,7 @@
<name>ping</name>
<ping>
<exchange>amq.direct</exchange>
+ <durable>true</durable>
<maximumQueueDepth>4235264</maximumQueueDepth> <!-- 4Mb -->
<maximumMessageSize>2117632</maximumMessageSize> <!-- 2Mb -->
<maximumMessageAge>600000</maximumMessageAge> <!-- 10 mins -->
@@ -97,8 +111,6 @@
<name>test</name>
<test>
<queues>
- <minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
- <maximumMessageCount>5000</maximumMessageCount>
<queue>
<name>queue</name>
<queue>
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/python-test.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/python-test.xml?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/python-test.xml (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/python-test.xml Mon Apr 21 07:03:35 2008
@@ -46,6 +46,8 @@
<arg value="2110"/>
<arg value="-m"/>
<arg value="2111"/>
+ <arg value="-s"/>
+ <arg value="../spec/amqp.0-9.no-wip.xml"/>
<classpath refid="maven.test.classpath"/>
<sysproperty key="QPID_HOME" value="${broker.dir}"/>
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java Mon Apr 21 07:03:35 2008
@@ -56,8 +56,9 @@
import org.apache.qpid.server.management.MBeanDescription;
import org.apache.qpid.server.management.ManagedBroker;
import org.apache.qpid.server.management.ManagedObject;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -175,7 +176,7 @@
ownerShortString = new AMQShortString(owner);
}
- queue = new AMQQueue(new AMQShortString(queueName), durable, ownerShortString, false, getVirtualHost());
+ queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString(queueName), durable, ownerShortString, false, getVirtualHost());
if (queue.isDurable() && !queue.isAutoDelete())
{
_messageStore.createQueue(queue);
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Mon Apr 21 07:03:35 2008
@@ -21,7 +21,6 @@
package org.apache.qpid.server;
import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.configuration.Configured;
import org.apache.qpid.framing.AMQShortString;
@@ -30,14 +29,21 @@
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.ack.UnacknowledgedMessage;
import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
import org.apache.qpid.server.configuration.Configurator;
-import org.apache.qpid.server.exchange.NoRouteException;
import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.exchange.NoRouteException;
+import org.apache.qpid.server.flow.FlowCreditManager;
+import org.apache.qpid.server.flow.Pre0_10CreditManager;
import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.server.queue.*;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.IncomingMessage;
+import org.apache.qpid.server.queue.MessageHandleFactory;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.txn.LocalTransactionalContext;
@@ -45,12 +51,11 @@
import org.apache.qpid.server.txn.TransactionalContext;
import java.util.Collection;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
public class AMQChannel
@@ -61,13 +66,8 @@
private final int _channelId;
- // private boolean _transactional;
-
- private long _prefetch_HighWaterMark;
- private long _prefetch_LowWaterMark;
-
- private long _prefetchSize;
+ private final Pre0_10CreditManager _creditManager = new Pre0_10CreditManager(0l,0l);
/**
* The delivery tag is unique per channel. This is pre-incremented before putting into the deliver frame so that
@@ -86,10 +86,11 @@
* been received by this channel. As the frames are received the message gets updated and once all frames have been
* received the message can then be routed.
*/
- private AMQMessage _currentMessage;
+ private IncomingMessage _currentMessage;
+
+ /** Maps from consumer tag to subscription instance. Allows us to unsubscribe from a queue. */
+ private final Map<AMQShortString, Subscription> _tag2SubscriptionMap = new HashMap<AMQShortString, Subscription>();
- /** Maps from consumer tag to queue instance. Allows us to unsubscribe from a queue. */
- private final Map<AMQShortString, AMQQueue> _consumerTag2QueueMap = new ConcurrentHashMap<AMQShortString, AMQQueue>();
private final MessageStore _messageStore;
@@ -97,7 +98,7 @@
private final AtomicBoolean _suspended = new AtomicBoolean(false);
- private TransactionalContext _txnContext, _nonTransactedContext;
+ private TransactionalContext _txnContext;
/**
* A context used by the message store enabling it to track context for a given channel even across thread
@@ -109,8 +110,6 @@
private MessageHandleFactory _messageHandleFactory = new MessageHandleFactory();
- private Set<Long> _browsedAcks = new HashSet<Long>();
-
// Why do we need this reference ? - ritchiem
private final AMQProtocolSession _session;
private boolean _closing;
@@ -118,7 +117,7 @@
@Configured(path = "advanced.enableJMSXUserID",
defaultValue = "false")
public boolean ENABLE_JMSXUserID;
-
+
public AMQChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
throws AMQException
@@ -129,18 +128,18 @@
_session = session;
_channelId = channelId;
_storeContext = new StoreContext("Session: " + session.getClientIdentifier() + "; channel: " + channelId);
- _prefetch_HighWaterMark = DEFAULT_PREFETCH;
- _prefetch_LowWaterMark = _prefetch_HighWaterMark / 2;
+
+
_messageStore = messageStore;
// by default the session is non-transactional
- _txnContext = new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages, _browsedAcks);
+ _txnContext = new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages);
}
/** Sets this channel to be part of a local transaction */
public void setLocalTransactional()
{
- _txnContext = new LocalTransactionalContext(_messageStore, _storeContext, _returnMessages);
+ _txnContext = new LocalTransactionalContext(this);
}
public boolean isTransactional()
@@ -156,55 +155,15 @@
return _channelId;
}
- public long getPrefetchCount()
+ public void setPublishFrame(MessagePublishInfo info, final Exchange e) throws AMQException
{
- return _prefetch_HighWaterMark;
- }
- public void setPrefetchCount(long prefetchCount)
- {
- _prefetch_HighWaterMark = prefetchCount;
- }
-
- public long getPrefetchSize()
- {
- return _prefetchSize;
- }
-
- public void setPrefetchSize(long prefetchSize)
- {
- _prefetchSize = prefetchSize;
- }
-
- public long getPrefetchLowMarkCount()
- {
- return _prefetch_LowWaterMark;
- }
-
- public void setPrefetchLowMarkCount(long prefetchCount)
- {
- _prefetch_LowWaterMark = prefetchCount;
- }
-
- public long getPrefetchHighMarkCount()
- {
- return _prefetch_HighWaterMark;
- }
-
- public void setPrefetchHighMarkCount(long prefetchCount)
- {
- _prefetch_HighWaterMark = prefetchCount;
- }
-
- public void setPublishFrame(MessagePublishInfo info, AMQProtocolSession publisher, final Exchange e) throws AMQException
- {
-
- _currentMessage = new AMQMessage(_messageStore.getNewMessageId(), info, _txnContext);
- _currentMessage.setPublisher(publisher);
+ _currentMessage = new IncomingMessage(_messageStore.getNewMessageId(), info, _txnContext, _session);
+ _currentMessage.setMessageStore(_messageStore);
_currentMessage.setExchange(e);
}
- public void publishContentHeader(ContentHeaderBody contentHeaderBody, AMQProtocolSession protocolSession)
+ public void publishContentHeader(ContentHeaderBody contentHeaderBody)
throws AMQException
{
if (_currentMessage == null)
@@ -215,7 +174,7 @@
{
if (_log.isDebugEnabled())
{
- _log.debug(debugIdentity() + "Content header received on channel " + _channelId);
+ _log.debug("Content header received on channel " + _channelId);
}
if (ENABLE_JMSXUserID)
@@ -225,25 +184,48 @@
//fixme: fudge for QPID-677
properties.getHeaders().keySet();
- properties.setUserId(protocolSession.getAuthorizedID().getName());
+ properties.setUserId(_session.getAuthorizedID().getName());
}
_currentMessage.setContentHeaderBody(contentHeaderBody);
+
_currentMessage.setExpiration();
routeCurrentMessage();
- _currentMessage.routingComplete(_messageStore, _storeContext, _messageHandleFactory);
- // check and deliver if header says body length is zero
- if (contentHeaderBody.bodySize == 0)
+ _currentMessage.routingComplete(_messageStore, _messageHandleFactory);
+
+ deliverCurrentMessageIfComplete();
+
+ }
+ }
+
+ private void deliverCurrentMessageIfComplete()
+ throws AMQException
+ {
+ // check and deliver if header says body length is zero
+ if (_currentMessage.allContentReceived())
+ {
+ try
{
- _txnContext.messageProcessed(protocolSession);
+ _currentMessage.deliverToQueues();
+ }
+ catch (NoRouteException e)
+ {
+ _returnMessages.add(e);
+ }
+ finally
+ {
+ // callback to allow the context to do any post message processing
+ // primary use is to allow message return processing in the non-tx case
+ _txnContext.messageProcessed(_session);
_currentMessage = null;
}
}
+
}
- public void publishContentBody(ContentBody contentBody, AMQProtocolSession protocolSession) throws AMQException
+ public void publishContentBody(ContentBody contentBody) throws AMQException
{
if (_currentMessage == null)
{
@@ -260,15 +242,11 @@
// returns true iff the message was delivered (i.e. if all data was
// received
- if (_currentMessage.addContentBodyFrame(_storeContext,
- protocolSession.getMethodRegistry().getProtocolVersionMethodConverter().convertToContentChunk(
- contentBody)))
- {
- // callback to allow the context to do any post message processing
- // primary use is to allow message return processing in the non-tx case
- _txnContext.messageProcessed(protocolSession);
- _currentMessage = null;
- }
+ _currentMessage.addContentBodyFrame(
+ _session.getMethodRegistry().getProtocolVersionMethodConverter().convertToContentChunk(
+ contentBody));
+
+ deliverCurrentMessageIfComplete();
}
catch (AMQException e)
{
@@ -287,6 +265,7 @@
}
catch (NoRouteException e)
{
+ //_currentMessage.incrementReference();
_returnMessages.add(e);
}
}
@@ -307,18 +286,17 @@
*
* @param tag the tag chosen by the client (if null, server will generate one)
* @param queue the queue to subscribe to
- * @param session the protocol session of the subscriber
- * @param noLocal Flag stopping own messages being receivied.
- * @param exclusive Flag requesting exclusive access to the queue
* @param acks Are acks enabled for this subscriber
* @param filters Filters to apply to this subscriber
*
+ * @param noLocal Flag stopping own messages being receivied.
+ * @param exclusive Flag requesting exclusive access to the queue
* @return the consumer tag. This is returned to the subscriber and used in subsequent unsubscribe requests
*
* @throws ConsumerTagNotUniqueException if the tag is not unique
* @throws AMQException if something goes wrong
*/
- public AMQShortString subscribeToQueue(AMQShortString tag, AMQQueue queue, AMQProtocolSession session, boolean acks,
+ public AMQShortString subscribeToQueue(AMQShortString tag, AMQQueue queue, boolean acks,
FieldTable filters, boolean noLocal, boolean exclusive) throws AMQException, ConsumerTagNotUniqueException
{
if (tag == null)
@@ -326,77 +304,65 @@
tag = new AMQShortString("sgen_" + getNextConsumerTag());
}
- if (_consumerTag2QueueMap.containsKey(tag))
+ if (_tag2SubscriptionMap.containsKey(tag))
{
throw new ConsumerTagNotUniqueException();
}
+ Subscription subscription =
+ SubscriptionFactoryImpl.INSTANCE.createSubscription(_channelId, _session, tag, acks, filters, noLocal, _creditManager);
+
+
+ // So to keep things straight we put before the call and catch all exceptions from the register and tidy up.
// We add before we register as the Async Delivery process may AutoClose the subscriber
// so calling _cT2QM.remove before we have done put which was after the register succeeded.
// So to keep things straight we put before the call and catch all exceptions from the register and tidy up.
- _consumerTag2QueueMap.put(tag, queue);
+
+ _tag2SubscriptionMap.put(tag, subscription);
try
{
- queue.registerProtocolSession(session, _channelId, tag, acks, filters, noLocal, exclusive);
+ queue.registerSubscription(subscription, exclusive);
}
catch (AMQException e)
{
- _consumerTag2QueueMap.remove(tag);
+ _tag2SubscriptionMap.remove(tag);
throw e;
}
-
return tag;
}
/**
* Unsubscribe a consumer from a queue.
- * @param session
* @param consumerTag
* @return true if the consumerTag had a mapped queue that could be unregistered.
* @throws AMQException
*/
- public boolean unsubscribeConsumer(AMQProtocolSession session, AMQShortString consumerTag) throws AMQException
+ public boolean unsubscribeConsumer(AMQShortString consumerTag) throws AMQException
{
- if (_log.isDebugEnabled())
- {
- _log.debug("Unacked Map Dump size:" + _unacknowledgedMessageMap.size());
- _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
- {
-
- public boolean callback(UnacknowledgedMessage message) throws AMQException
- {
- _log.debug(message);
-
- return true;
- }
- public void visitComplete()
- {
- }
- });
- }
-
- AMQQueue q = _consumerTag2QueueMap.remove(consumerTag);
- if (q != null)
+ Subscription sub = _tag2SubscriptionMap.remove(consumerTag);
+ if (sub != null)
{
- q.unregisterProtocolSession(session, _channelId, consumerTag);
+ sub.getQueue().unregisterSubscription(sub);
return true;
}
+ else
+ {
+ _log.warn("Attempt to unsubscribe consumer with tag '"+consumerTag+"' which is not registered.");
+ }
return false;
}
/**
* Called from the protocol session to close this channel and clean up. T
*
- * @param session The session to close
- *
* @throws AMQException if there is an error during closure
*/
- public void close(AMQProtocolSession session) throws AMQException
+ public void close() throws AMQException
{
_txnContext.rollback();
- unsubscribeAllConsumers(session);
+ unsubscribeAllConsumers();
try
{
requeue();
@@ -414,11 +380,11 @@
_closing = closing;
}
- private void unsubscribeAllConsumers(AMQProtocolSession session) throws AMQException
+ private void unsubscribeAllConsumers() throws AMQException
{
if (_log.isInfoEnabled())
{
- if (!_consumerTag2QueueMap.isEmpty())
+ if (!_tag2SubscriptionMap.isEmpty())
{
_log.info("Unsubscribing all consumers on channel " + toString());
}
@@ -428,17 +394,19 @@
}
}
- for (Map.Entry<AMQShortString, AMQQueue> me : _consumerTag2QueueMap.entrySet())
+ for (Map.Entry<AMQShortString, Subscription> me : _tag2SubscriptionMap.entrySet())
{
if (_log.isInfoEnabled())
{
_log.info("Unsubscribing consumer '" + me.getKey() + "' on channel " + toString());
}
- me.getValue().unregisterProtocolSession(session, _channelId, me.getKey());
+ Subscription sub = me.getValue();
+
+ sub.getQueue().unregisterSubscription(sub);
}
- _consumerTag2QueueMap.clear();
+ _tag2SubscriptionMap.clear();
}
/**
@@ -447,9 +415,9 @@
* @param entry the record of the message on the queue that was delivered
* @param deliveryTag the delivery tag used when delivering the message (see protocol spec for description of the
* delivery tag)
- * @param consumerTag The tag for the consumer that is to acknowledge this message.
+ * @param subscription The consumer that is to acknowledge this message.
*/
- public void addUnacknowledgedMessage(QueueEntry entry, long deliveryTag, AMQShortString consumerTag)
+ public void addUnacknowledgedMessage(QueueEntry entry, long deliveryTag, Subscription subscription)
{
if (_log.isDebugEnabled())
{
@@ -462,15 +430,14 @@
if (_log.isDebugEnabled())
{
_log.debug(debugIdentity() + " Adding unacked message(" + entry.getMessage().toString() + " DT:" + deliveryTag
- + ") with a queue(" + entry.getQueue() + ") for " + consumerTag);
+ + ") with a queue(" + entry.getQueue() + ") for " + subscription);
}
}
}
synchronized (_unacknowledgedMessageMap.getLock())
{
- _unacknowledgedMessageMap.add(deliveryTag, new UnacknowledgedMessage(entry, consumerTag, deliveryTag));
- checkSuspension();
+ _unacknowledgedMessageMap.add(deliveryTag, entry);
}
}
@@ -490,7 +457,7 @@
public void requeue() throws AMQException
{
// we must create a new map since all the messages will get a new delivery tag when they are redelivered
- Collection<UnacknowledgedMessage> messagesToBeDelivered = _unacknowledgedMessageMap.cancelAllMessages();
+ Collection<QueueEntry> messagesToBeDelivered = _unacknowledgedMessageMap.cancelAllMessages();
// Deliver these messages out of the transaction as their delivery was never
// part of the transaction only the receive.
@@ -505,13 +472,9 @@
if (!(_txnContext instanceof NonTransactionalContext))
{
- // if (_nonTransactedContext == null)
- {
- _nonTransactedContext =
- new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages, _browsedAcks);
- }
- deliveryContext = _nonTransactedContext;
+ deliveryContext =
+ new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages);
}
else
{
@@ -519,23 +482,27 @@
}
}
- for (UnacknowledgedMessage unacked : messagesToBeDelivered)
+ for (QueueEntry unacked : messagesToBeDelivered)
{
if (!unacked.isQueueDeleted())
{
- // Ensure message is released for redelivery
- unacked.entry.release();
-
// Mark message redelivered
unacked.getMessage().setRedelivered(true);
+ // Ensure message is released for redelivery
+ unacked.release();
+
// Deliver Message
- deliveryContext.deliver(unacked.entry, false);
+ deliveryContext.requeue(unacked);
// Should we allow access To the DM to directy deliver the message?
// As we don't need to check for Consumers or worry about incrementing the message count?
// unacked.queue.getDeliveryManager().deliver(_storeContext, unacked.queue.getName(), unacked.message, false);
}
+ else
+ {
+ unacked.discard(_storeContext);
+ }
}
}
@@ -549,32 +516,29 @@
*/
public void requeue(long deliveryTag) throws AMQException
{
- UnacknowledgedMessage unacked = _unacknowledgedMessageMap.remove(deliveryTag);
+ QueueEntry unacked = _unacknowledgedMessageMap.remove(deliveryTag);
if (unacked != null)
{
+ // Mark message redelivered
+ unacked.getMessage().setRedelivered(true);
// Ensure message is released for redelivery
if (!unacked.isQueueDeleted())
{
- unacked.entry.release();
+ unacked.release();
}
- // Mark message redelivered
- unacked.getMessage().setRedelivered(true);
// Deliver these messages out of the transaction as their delivery was never
// part of the transaction only the receive.
TransactionalContext deliveryContext;
if (!(_txnContext instanceof NonTransactionalContext))
{
- // if (_nonTransactedContext == null)
- {
- _nonTransactedContext =
- new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages, _browsedAcks);
- }
- deliveryContext = _nonTransactedContext;
+ deliveryContext =
+ new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages);
+
}
else
{
@@ -584,7 +548,7 @@
if (!unacked.isQueueDeleted())
{
// Redeliver the messages to the front of the queue
- deliveryContext.deliver(unacked.entry, true);
+ deliveryContext.requeue(unacked);
// Deliver increments the message count but we have already deliverted this once so don't increment it again
// this was because deliver did an increment changed this.
}
@@ -592,11 +556,8 @@
{
_log.warn(System.identityHashCode(this) + " Requested requeue of message(" + unacked.getMessage().debugIdentity()
+ "):" + deliveryTag + " but no queue defined and no DeadLetter queue so DROPPING message.");
- // _log.error("Requested requeue of message:" + deliveryTag +
- // " but no queue defined using DeadLetter queue:" + getDeadLetterQueue());
- //
- // deliveryContext.deliver(unacked.message, getDeadLetterQueue(), false);
- //
+
+ unacked.discard(_storeContext);
}
}
else
@@ -604,25 +565,6 @@
_log.warn("Requested requeue of message:" + deliveryTag + " but no such delivery tag exists."
+ _unacknowledgedMessageMap.size());
- if (_log.isDebugEnabled())
- {
- _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
- {
- int count = 0;
-
- public boolean callback(UnacknowledgedMessage message) throws AMQException
- {
- _log.debug(
- (count++) + ": (" + message.getMessage().debugIdentity() + ")" + "[" + message.deliveryTag + "]");
-
- return false; // Continue
- }
-
- public void visitComplete()
- {
- }
- });
- }
}
}
@@ -636,8 +578,10 @@
*/
public void resend(final boolean requeue) throws AMQException
{
- final List<UnacknowledgedMessage> msgToRequeue = new LinkedList<UnacknowledgedMessage>();
- final List<UnacknowledgedMessage> msgToResend = new LinkedList<UnacknowledgedMessage>();
+
+
+ final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+ final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
if (_log.isDebugEnabled())
{
@@ -647,23 +591,25 @@
// Process the Unacked-Map.
// Marking messages who still have a consumer for to be resent
// and those that don't to be requeued.
+
_unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
{
- public boolean callback(UnacknowledgedMessage message) throws AMQException
+ public boolean callback(final long deliveryTag, QueueEntry message) throws AMQException
{
- AMQShortString consumerTag = message.consumerTag;
+
AMQMessage msg = message.getMessage();
msg.setRedelivered(true);
- if (consumerTag != null)
+ final Subscription subscription = message.getDeliveredSubscription();
+ if (subscription != null)
{
// Consumer exists
- if (_consumerTag2QueueMap.containsKey(consumerTag))
+ if (!subscription.isClosed())
{
- msgToResend.add(message);
+ msgToResend.put(deliveryTag, message);
}
else // consumer has gone
{
- msgToRequeue.add(message);
+ msgToRequeue.put(deliveryTag, message);
}
}
else
@@ -675,7 +621,7 @@
{
if (requeue)
{
- msgToRequeue.add(message);
+ msgToRequeue.put(deliveryTag, message);
}
else
{
@@ -684,7 +630,8 @@
}
else
{
- _log.info("Message.queue is null and no DeadLetter Queue so dropping message:" + message);
+ message.discard(_storeContext);
+ _log.warn("Message.queue is null and no DeadLetter Queue so dropping message:" + message);
}
}
@@ -697,6 +644,8 @@
}
});
+ _unacknowledgedMessageMap.clear();
+
// Process Messages to Resend
if (_log.isDebugEnabled())
{
@@ -710,9 +659,15 @@
}
}
- for (UnacknowledgedMessage message : msgToResend)
+ for (Map.Entry<Long, QueueEntry> entry : msgToResend.entrySet())
{
+ QueueEntry message = entry.getValue();
+ long deliveryTag = entry.getKey();
+
+
+
AMQMessage msg = message.getMessage();
+ AMQQueue queue = message.getQueue();
// Our Java Client will always suspend the channel when resending!
// If the client has requested the messages be resent then it is
@@ -727,46 +682,20 @@
// else
// {
// release to allow it to be delivered
- message.entry.release();
// Without any details from the client about what has been processed we have to mark
// all messages in the unacked map as redelivered.
msg.setRedelivered(true);
- Subscription sub = message.entry.getDeliveredSubscription();
+ Subscription sub = message.getDeliveredSubscription();
if (sub != null)
{
- // Get the lock so we can tell if the sub scription has closed.
- // will stop delivery to this subscription until the lock is released.
- // note: this approach would allow the use of a single queue if the
- // PreDeliveryQueue would allow head additions.
- // In the Java Qpid client we are suspended whilst doing this so it is all rather Mute..
- // needs guidance from AMQP WG Model SIG
- synchronized (sub.getSendLock())
+
+ if(!queue.resend(message, sub))
{
- if (sub.isClosed())
- {
- if (_log.isDebugEnabled())
- {
- _log.debug("Subscription(" + System.identityHashCode(sub)
- + ") closed during resend so requeuing message");
- }
- // move this message to requeue
- msgToRequeue.add(message);
- }
- else
- {
- if (_log.isDebugEnabled())
- {
- _log.debug("Requeuing " + msg.debugIdentity() + " for resend via sub:"
- + System.identityHashCode(sub));
- }
-
- sub.addToResendQueue(message.entry);
- _unacknowledgedMessageMap.remove(message.deliveryTag);
- }
- } // sync(sub.getSendLock)
+ msgToRequeue.put(deliveryTag, message);
+ }
}
else
{
@@ -777,7 +706,7 @@
+ ")to prevent loss");
}
// move this message to requeue
- msgToRequeue.add(message);
+ msgToRequeue.put(deliveryTag, message);
}
} // for all messages
// } else !isSuspend
@@ -795,13 +724,9 @@
TransactionalContext deliveryContext;
if (!(_txnContext instanceof NonTransactionalContext))
{
- if (_nonTransactedContext == null)
- {
- _nonTransactedContext =
- new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages, _browsedAcks);
- }
- deliveryContext = _nonTransactedContext;
+ deliveryContext =
+ new NonTransactionalContext(_messageStore, _storeContext, this, _returnMessages);
}
else
{
@@ -809,14 +734,17 @@
}
// Process Messages to Requeue at the front of the queue
- for (UnacknowledgedMessage message : msgToRequeue)
+ for (Map.Entry<Long, QueueEntry> entry : msgToRequeue.entrySet())
{
- message.entry.release();
- message.entry.setRedelivered(true);
+ QueueEntry message = entry.getValue();
+ long deliveryTag = entry.getKey();
+
+ message.release();
+ message.setRedelivered(true);
- deliveryContext.deliver(message.entry, true);
+ deliveryContext.requeue(message);
- _unacknowledgedMessageMap.remove(message.deliveryTag);
+ _unacknowledgedMessageMap.remove(deliveryTag);
}
}
@@ -827,38 +755,47 @@
*
* @param queue the queue that has been deleted
*
- * @throws org.apache.qpid.AMQException if there is an error processing the unacked messages
*/
- public void queueDeleted(final AMQQueue queue) throws AMQException
+ /* public void queueDeleted(final AMQQueue queue)
{
- _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
+ try
{
- public boolean callback(UnacknowledgedMessage message) throws AMQException
+ _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
{
- if (message.getQueue() == queue)
+ public boolean callback(UnacknowledgedMessage message)
{
- try
+ if (message.getQueue() == queue)
{
- message.discard(_storeContext);
- message.setQueueDeleted(true);
+ try
+ {
+ message.discard(_storeContext);
+ message.setQueueDeleted(true);
+ }
+ catch (AMQException e)
+ {
+ _log.error(
+ "Error decrementing ref count on message " + message.getMessage().getMessageId() + ": " + e, e);
+ throw new RuntimeException(e);
+ }
}
- catch (AMQException e)
- {
- _log.error(
- "Error decrementing ref count on message " + message.getMessage().getMessageId() + ": " + e, e);
- }
+
+ return false;
}
- return false;
- }
+ public void visitComplete()
+ {
+ }
+ });
+ }
+ catch (AMQException e)
+ {
+ _log.error("Unexpected Error while handling deletion of queue", e);
+ throw new RuntimeException(e);
+ }
- public void visitComplete()
- {
- }
- });
}
-
+*/
/**
* Acknowledge one or more messages.
*
@@ -886,7 +823,6 @@
}
- checkSuspension();
}
/**
@@ -899,26 +835,11 @@
return _unacknowledgedMessageMap;
}
- private void checkSuspension()
- {
- boolean suspend;
-
- suspend =
- ((_prefetch_HighWaterMark != 0) && (_unacknowledgedMessageMap.size() >= _prefetch_HighWaterMark))
- || ((_prefetchSize != 0) && (_prefetchSize < _unacknowledgedMessageMap.getUnacknowledgeBytes()));
- setSuspended(suspend);
- }
public void setSuspended(boolean suspended)
{
- boolean isSuspended = _suspended.get();
- if (isSuspended && !suspended)
- {
- // Continue being suspended if we are above the _prefetch_LowWaterMark
- suspended = _unacknowledgedMessageMap.size() > _prefetch_LowWaterMark;
- }
boolean wasSuspended = _suspended.getAndSet(suspended);
if (wasSuspended != suspended)
@@ -927,9 +848,9 @@
{
_log.debug("Unsuspending channel " + this);
// may need to deliver queued messages
- for (AMQQueue q : _consumerTag2QueueMap.values())
+ for (Subscription s : _tag2SubscriptionMap.values())
{
- q.deliverAsync();
+ s.getQueue().deliverAsync();
}
}
else
@@ -961,12 +882,7 @@
public String toString()
{
- StringBuilder sb = new StringBuilder(30);
- sb.append("Channel: id ").append(_channelId).append(", transaction mode: ").append(isTransactional());
- sb.append(", prefetch marks: ").append(_prefetch_LowWaterMark);
- sb.append("/").append(_prefetch_HighWaterMark);
-
- return sb.toString();
+ return "["+_session.toString()+":"+_channelId+"]";
}
public void setDefaultQueue(AMQQueue queue)
@@ -984,14 +900,14 @@
return _storeContext;
}
- public void processReturns(AMQProtocolSession session) throws AMQException
+ public void processReturns() throws AMQException
{
if (!_returnMessages.isEmpty())
{
for (RequiredDeliveryException bouncedMessage : _returnMessages)
{
AMQMessage message = bouncedMessage.getAMQMessage();
- session.getProtocolOutputConverter().writeReturn(message, _channelId, bouncedMessage.getReplyCode().getCode(),
+ _session.getProtocolOutputConverter().writeReturn(message, _channelId, bouncedMessage.getReplyCode().getCode(),
new AMQShortString(bouncedMessage.getMessage()));
message.decrementReference(_storeContext);
@@ -1001,40 +917,39 @@
}
}
- public boolean wouldSuspend(AMQMessage msg)
+
+ public TransactionalContext getTransactionalContext()
{
- if (isSuspended())
- {
- return true;
- }
- else
- {
- boolean willSuspend =
- ((_prefetch_HighWaterMark != 0) && ((_unacknowledgedMessageMap.size() + 1) > _prefetch_HighWaterMark));
- if (!willSuspend)
- {
- final long unackedSize = _unacknowledgedMessageMap.getUnacknowledgeBytes();
+ return _txnContext;
+ }
- willSuspend = (_prefetchSize != 0) && (unackedSize != 0) && (_prefetchSize < (msg.getSize() + unackedSize));
- }
+ public boolean isClosing()
+ {
+ return _closing;
+ }
- if (willSuspend)
- {
- setSuspended(true);
- }
+ public AMQProtocolSession getProtocolSession()
+ {
+ return _session;
+ }
- return willSuspend;
- }
+ public FlowCreditManager getCreditManager()
+ {
+ return _creditManager;
+ }
+ public void setCredit(final long prefetchSize, final int prefetchCount)
+ {
+ _creditManager.setCreditLimits(prefetchSize, prefetchCount);
}
- public TransactionalContext getTransactionalContext()
+ public List<RequiredDeliveryException> getReturnMessages()
{
- return _txnContext;
+ return _returnMessages;
}
- public boolean isClosing()
+ public MessageStore getMessageStore()
{
- return _closing;
+ return _messageStore;
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/RequiredDeliveryException.java Mon Apr 21 07:03:35 2008
@@ -39,19 +39,30 @@
*/
public abstract class RequiredDeliveryException extends AMQException
{
- private final AMQMessage _amqMessage;
+ private AMQMessage _amqMessage;
public RequiredDeliveryException(String message, AMQMessage payload)
{
super(message);
+ setMessage(payload);
+ }
+
+
+ public RequiredDeliveryException(String message)
+ {
+ super(message);
+ }
+
+ public void setMessage(final AMQMessage payload)
+ {
+
// Increment the reference as this message is in the routing phase
// and so will have the ref decremented as routing fails.
// we need to keep this message around so we can return it in the
// handler. So increment here.
_amqMessage = payload.takeReference();
- // payload.incrementReference();
}
public AMQMessage getAMQMessage()
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/TxAck.java Mon Apr 21 07:03:35 2008
@@ -22,10 +22,13 @@
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.txn.TxnOp;
+import org.apache.qpid.server.queue.QueueEntry;
/**
* A TxnOp implementation for handling accumulated acks
@@ -33,7 +36,7 @@
public class TxAck implements TxnOp
{
private final UnacknowledgedMessageMap _map;
- private final List <UnacknowledgedMessage> _unacked = new LinkedList<UnacknowledgedMessage>();
+ private final Map<Long, QueueEntry> _unacked = new HashMap<Long,QueueEntry>();
private final List<Long> _individual = new LinkedList<Long>();
private long _deliveryTag;
private boolean _multiple;
@@ -47,6 +50,7 @@
{
if (!multiple)
{
+
//have acked a single message that is not part of
//the previously acked region so record
//individually
@@ -85,7 +89,7 @@
{
//if any of the messages in unacked are persistent the txn
//buffer must be marked as persistent:
- for (UnacknowledgedMessage msg : _unacked)
+ for (QueueEntry msg : _unacked.values())
{
if (msg.getMessage().isPersistent())
{
@@ -98,7 +102,7 @@
public void prepare(StoreContext storeContext) throws AMQException
{
//make persistent changes, i.e. dequeue and decrementReference
- for (UnacknowledgedMessage msg : _unacked)
+ for (QueueEntry msg : _unacked.values())
{
//Message has been ack so discard it. This will dequeue and decrement the reference.
msg.discard(storeContext);
@@ -112,7 +116,7 @@
//in memory counter) so if we failed in prepare for full
//txn, this op will have to compensate by fixing the count
//in memory (persistent changes will be rolled back by store)
- for (UnacknowledgedMessage msg : _unacked)
+ for (QueueEntry msg : _unacked.values())
{
msg.getMessage().takeReference();
}
@@ -121,7 +125,7 @@
public void commit(StoreContext storeContext)
{
//remove the unacked messages from the channels map
- _map.remove(_unacked);
+ _map.remove(_unacked);
}
public void rollback(StoreContext storeContext)
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMap.java Mon Apr 21 07:03:35 2008
@@ -23,20 +23,22 @@
import java.util.Collection;
import java.util.List;
import java.util.Set;
+import java.util.Map;
import org.apache.qpid.AMQException;
import org.apache.qpid.server.txn.TransactionalContext;
+import org.apache.qpid.server.queue.QueueEntry;
public interface UnacknowledgedMessageMap
{
public interface Visitor
{
/**
- * @param message the message being iterated over
- * @return true to stop iteration, false to continue
+ * @param deliveryTag
+ *@param message the message being iterated over @return true to stop iteration, false to continue
* @throws AMQException
*/
- boolean callback(UnacknowledgedMessage message) throws AMQException;
+ boolean callback(final long deliveryTag, QueueEntry message) throws AMQException;
void visitComplete();
}
@@ -45,19 +47,19 @@
Object getLock();
- void add(long deliveryTag, UnacknowledgedMessage message);
+ void add(long deliveryTag, QueueEntry message);
- void collect(long deliveryTag, boolean multiple, List<UnacknowledgedMessage> msgs);
+ void collect(long deliveryTag, boolean multiple, Map<Long, QueueEntry> msgs);
boolean contains(long deliveryTag) throws AMQException;
- void remove(List<UnacknowledgedMessage> msgs);
+ void remove(Map<Long,QueueEntry> msgs);
- UnacknowledgedMessage remove(long deliveryTag);
+ QueueEntry remove(long deliveryTag);
- void drainTo(Collection<UnacknowledgedMessage> destination, long deliveryTag) throws AMQException;
+ void drainTo(Collection<QueueEntry> destination, long deliveryTag) throws AMQException;
- Collection<UnacknowledgedMessage> cancelAllMessages();
+ Collection<QueueEntry> cancelAllMessages();
void acknowledgeMessage(long deliveryTag, boolean multiple, TransactionalContext txnContext) throws AMQException;
@@ -65,7 +67,7 @@
void clear();
- UnacknowledgedMessage get(long deliveryTag);
+ QueueEntry get(long deliveryTag);
/**
* Get the set of delivery tags that are outstanding.
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/ack/UnacknowledgedMessageMapImpl.java Mon Apr 21 07:03:35 2008
@@ -31,6 +31,7 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.txn.TransactionalContext;
public class UnacknowledgedMessageMapImpl implements UnacknowledgedMessageMap
@@ -39,7 +40,7 @@
private long _unackedSize;
- private Map<Long, UnacknowledgedMessage> _map;
+ private Map<Long, QueueEntry> _map;
private long _lastDeliveryTag;
@@ -48,16 +49,10 @@
public UnacknowledgedMessageMapImpl(int prefetchLimit)
{
_prefetchLimit = prefetchLimit;
- _map = new LinkedHashMap<Long, UnacknowledgedMessage>(prefetchLimit);
+ _map = new LinkedHashMap<Long, QueueEntry>(prefetchLimit);
}
- /*public UnacknowledgedMessageMapImpl(Object lock, Map<Long, UnacknowledgedMessage> map)
- {
- _lock = lock;
- _map = map;
- } */
-
- public void collect(long deliveryTag, boolean multiple, List<UnacknowledgedMessage> msgs)
+ public void collect(long deliveryTag, boolean multiple, Map<Long, QueueEntry> msgs)
{
if (multiple)
{
@@ -65,7 +60,7 @@
}
else
{
- msgs.add(get(deliveryTag));
+ msgs.put(deliveryTag, get(deliveryTag));
}
}
@@ -78,26 +73,27 @@
}
}
- public void remove(List<UnacknowledgedMessage> msgs)
+ public void remove(Map<Long,QueueEntry> msgs)
{
synchronized (_lock)
{
- for (UnacknowledgedMessage msg : msgs)
+ for (Long deliveryTag : msgs.keySet())
{
- remove(msg.deliveryTag);
+ remove(deliveryTag);
}
}
}
- public UnacknowledgedMessage remove(long deliveryTag)
+ public QueueEntry remove(long deliveryTag)
{
synchronized (_lock)
{
- UnacknowledgedMessage message = _map.remove(deliveryTag);
+ QueueEntry message = _map.remove(deliveryTag);
if(message != null)
{
_unackedSize -= message.getMessage().getSize();
+ message.restoreCredit();
}
return message;
@@ -108,10 +104,10 @@
{
synchronized (_lock)
{
- Collection<UnacknowledgedMessage> currentEntries = _map.values();
- for (UnacknowledgedMessage msg : currentEntries)
+ Set<Map.Entry<Long, QueueEntry>> currentEntries = _map.entrySet();
+ for (Map.Entry<Long, QueueEntry> entry : currentEntries)
{
- visitor.callback(msg);
+ visitor.callback(entry.getKey().longValue(), entry.getValue());
}
visitor.visitComplete();
}
@@ -122,7 +118,7 @@
return _lock;
}
- public void add(long deliveryTag, UnacknowledgedMessage message)
+ public void add(long deliveryTag, QueueEntry message)
{
synchronized (_lock)
{
@@ -132,12 +128,12 @@
}
}
- public Collection<UnacknowledgedMessage> cancelAllMessages()
+ public Collection<QueueEntry> cancelAllMessages()
{
synchronized (_lock)
{
- Collection<UnacknowledgedMessage> currentEntries = _map.values();
- _map = new LinkedHashMap<Long, UnacknowledgedMessage>(_prefetchLimit);
+ Collection<QueueEntry> currentEntries = _map.values();
+ _map = new LinkedHashMap<Long, QueueEntry>(_prefetchLimit);
_unackedSize = 0l;
return currentEntries;
}
@@ -169,14 +165,14 @@
}
}
- public void drainTo(Collection<UnacknowledgedMessage> destination, long deliveryTag) throws AMQException
+ public void drainTo(Collection<QueueEntry> destination, long deliveryTag) throws AMQException
{
synchronized (_lock)
{
- Iterator<Map.Entry<Long, UnacknowledgedMessage>> it = _map.entrySet().iterator();
+ Iterator<Map.Entry<Long, QueueEntry>> it = _map.entrySet().iterator();
while (it.hasNext())
{
- Map.Entry<Long, UnacknowledgedMessage> unacked = it.next();
+ Map.Entry<Long, QueueEntry> unacked = it.next();
if (unacked.getKey() > deliveryTag)
{
@@ -184,10 +180,13 @@
throw new AMQException("UnacknowledgedMessageMap is out of order:" + unacked.getKey() +
" When deliveryTag is:" + deliveryTag + "ES:" + _map.entrySet().toString());
}
-
it.remove();
+
_unackedSize -= unacked.getValue().getMessage().getSize();
+ unacked.getValue().restoreCredit();
+
+
destination.add(unacked.getValue());
if (unacked.getKey() == deliveryTag)
{
@@ -197,7 +196,7 @@
}
}
- public UnacknowledgedMessage get(long key)
+ public QueueEntry get(long key)
{
synchronized (_lock)
{
@@ -213,13 +212,13 @@
}
}
- private void collect(long key, List<UnacknowledgedMessage> msgs)
+ private void collect(long key, Map<Long, QueueEntry> msgs)
{
synchronized (_lock)
{
- for (Map.Entry<Long, UnacknowledgedMessage> entry : _map.entrySet())
+ for (Map.Entry<Long, QueueEntry> entry : _map.entrySet())
{
- msgs.add(entry.getValue());
+ msgs.put(entry.getKey(),entry.getValue());
if (entry.getKey() == key)
{
break;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/configuration/VirtualHostConfiguration.java Mon Apr 21 07:03:35 2008
@@ -35,6 +35,7 @@
import org.apache.qpid.server.exchange.ExchangeFactory;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -177,7 +178,7 @@
boolean autodelete = queueConfiguration.getBoolean("autodelete", false);
String owner = queueConfiguration.getString("owner", null);
- queue = new AMQQueue(queueName,
+ queue = AMQQueueFactory.createAMQQueueImpl(queueName,
durable,
owner == null ? null : new AMQShortString(owner) /* These queues will have no owner */,
autodelete /* Therefore autodelete makes no sence */, virtualHost);
@@ -221,7 +222,7 @@
AMQShortString routingKey = new AMQShortString(String.valueOf(routingKeyNameObj));
- queue.bind(routingKey, null, exchange);
+ queue.bind(exchange, routingKey, null);
_logger.info("Queue '" + queue.getName() + "' bound to exchange:" + exchangeName + " RK:'" + routingKey + "'");
@@ -229,7 +230,7 @@
if(exchange != virtualHost.getExchangeRegistry().getDefaultExchange())
{
- queue.bind(queue.getName(), null, virtualHost.getExchangeRegistry().getDefaultExchange());
+ queue.bind(virtualHost.getExchangeRegistry().getDefaultExchange(), queue.getName(), null);
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java Mon Apr 21 07:03:35 2008
@@ -38,6 +38,7 @@
import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.management.ManagedObjectRegistry;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueueImpl;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.virtualhost.VirtualHost;
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeFactory.java Mon Apr 21 07:03:35 2008
@@ -43,8 +43,8 @@
public DefaultExchangeFactory(VirtualHost host)
{
_host = host;
- registerExchangeType(DestNameExchange.TYPE);
- registerExchangeType(DestWildExchange.TYPE);
+ registerExchangeType(DirectExchange.TYPE);
+ registerExchangeType(TopicExchange.TYPE);
registerExchangeType(HeadersExchange.TYPE);
registerExchangeType(FanoutExchange.TYPE);
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/exchange/DefaultExchangeRegistry.java Mon Apr 21 07:03:35 2008
@@ -25,6 +25,7 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.protocol.ExchangeInitialiser;
import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.IncomingMessage;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -121,9 +122,9 @@
* @param payload
* @throws AMQException if something goes wrong delivering data
*/
- public void routeContent(AMQMessage payload) throws AMQException
+ public void routeContent(IncomingMessage payload) throws AMQException
{
- final AMQShortString exchange = payload.getMessagePublishInfo().getExchange();
+ final AMQShortString exchange = payload.getExchange();
final Exchange exch = getExchange(exchange);
// there is a small window of opportunity for the exchange to be deleted in between
// the BasicPublish being received (where the exchange is validated) and the final